From 5858041e8d629794e1a7c5a48834d1695e698949 Mon Sep 17 00:00:00 2001 From: Marco Ciavarella Date: Thu, 10 Sep 2026 14:34:49 +0200 Subject: [PATCH 1/4] Add file exploration contract and results, with unit tests --- .../src/exmergo_dex_core/files/__init__.py | 20 + .../src/exmergo_dex_core/files/contract.py | 813 ++++++++++++++++++ .../src/exmergo_dex_core/files/results.py | 488 +++++++++++ .../tests/files/test_file_contract.py | 469 ++++++++++ .../dex-core/tests/files/test_file_results.py | 623 ++++++++++++++ 5 files changed, 2413 insertions(+) create mode 100644 packages/dex-core/src/exmergo_dex_core/files/__init__.py create mode 100644 packages/dex-core/src/exmergo_dex_core/files/contract.py create mode 100644 packages/dex-core/src/exmergo_dex_core/files/results.py create mode 100644 packages/dex-core/tests/files/test_file_contract.py create mode 100644 packages/dex-core/tests/files/test_file_results.py diff --git a/packages/dex-core/src/exmergo_dex_core/files/__init__.py b/packages/dex-core/src/exmergo_dex_core/files/__init__.py new file mode 100644 index 00000000..26cbf381 --- /dev/null +++ b/packages/dex-core/src/exmergo_dex_core/files/__init__.py @@ -0,0 +1,20 @@ +"""File exploration: what a document collection holds and what is already known +about processing it. + +A file collection (a BigQuery object table, a Snowflake directory table, a +Databricks volume or manifest) is an index of files in object storage, and the +evidence worth having about it is usually a table some earlier pipeline +materialized by running a document parser. This package is the contract both are +read through. ``contract`` holds the fixed vocabularies, the optional interfaces a +connector may implement, the request types, and the capability model; ``results`` +holds the aggregate types that are the only thing a file operation returns. + +Two constraints shape everything here. Document content never leaves the +warehouse: every content computation is a warehouse expression, and the result +types have no field that could hold a document body, an extracted value, a file +path, or a provider's error text. And nothing here invokes document processing: +dex assesses results that already exist, because a new processing charge would +need a provider-enforced hard spend cap that no supported path offers. +""" + +from __future__ import annotations diff --git a/packages/dex-core/src/exmergo_dex_core/files/contract.py b/packages/dex-core/src/exmergo_dex_core/files/contract.py new file mode 100644 index 00000000..224f8cca --- /dev/null +++ b/packages/dex-core/src/exmergo_dex_core/files/contract.py @@ -0,0 +1,813 @@ +"""The file-exploration contract: vocabulary, interfaces, requests, capabilities. + +**Optional, and beside the warehouse adapter rather than inside it.** The +:class:`~..adapters.base.Adapter` protocol models tables and columns, and every +connector implements it. File exploration is something only some connectors can +do with a bounded, warehouse-side path to file metadata, so it lives in its own +protocols here. A connector that has nothing to say about files implements +nothing, and :func:`file_capabilities` reports that as a named limitation. +Widening ``Adapter`` instead would have demoted every host-supplied adapter that +had not grown the new members, and made every connector carry placeholder +document operations. + +**Tiers, checked structurally, never declared by a flag**, the same idiom as the +project seam in :mod:`..adapters.project`:: + + FileCollectionSource file_collection_inventory() -- metadata aggregation + DiscoveringFileSource list_file_collections() -- beside it, optional + FileResultSource + source_sample_sql() -- result assessment + + run_file_aggregate() + + quote_identifier() + +``isinstance(adapter, FileResultSource)`` is either true or it is not, so a +connector cannot claim a capability it does not implement. Discovery sits beside +the first tier rather than inside it because a source can aggregate a collection +it was explicitly handed (a bound manifest table) without being able to list +collections at all. + +**Every category is a fixed vocabulary.** Formats, statuses, limitations, and +unavailability reasons are enums, and anything a provider reports outside them is +counted under ``unknown`` or ``other`` rather than passed through. A category +that echoed provider strings would be a channel for whatever the provider put in +them, and a document parser's status field is where extracted text and error +bodies end up. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +from pydantic import BaseModel, ConfigDict, computed_field, model_validator + +from ..errors import RequestError + +if TYPE_CHECKING: + from .results import CollectionInventory, CollectionSummary, Unavailable + +__all__ = [ + "CONTENT_TYPE_FORMATS", + "DIAGNOSTIC_BIN_EDGES", + "FAMILY_FORMATS", + "LIMITATION_TEXT", + "RESULT_FORMATS", + "SAMPLE_CEILING", + "SAMPLE_DEFAULT", + "SHORTLIST_DEFAULT", + "Capability", + "CapabilityLimitation", + "CollectionKind", + "Diagnostic", + "DiscoveringFileSource", + "DocumentFamily", + "FileCapabilities", + "FileCollectionSource", + "FileFormat", + "FileResultSource", + "InventoryRequest", + "Limitation", + "MeasureKind", + "MetadataField", + "NativeProcessing", + "ProcessingStatus", + "ProfileRequest", + "ResultBinding", + "ResultFormat", + "ResultFormatName", + "RowDiagnostics", + "SamplingMethod", + "UnavailableReason", + "file_capabilities", + "format_for_content_type", +] + + +# --- Vocabulary -------------------------------------------------------------- + + +class DocumentFamily(str, Enum): + """A document family an assessment can be narrowed to. + + ``scanned_image`` denotes a document-image input (see + :data:`FAMILY_FORMATS`). It does not assert that dex verified the image + depicts a scanned document: nothing here looks at an image. + """ + + PDF = "pdf" + SCANNED_IMAGE = "scanned_image" + + +class FileFormat(str, Enum): + """The fixed format buckets a collection's files are counted into. + + A bucket is *reported metadata*: the content type the storage layer + records, never an inspection of the bytes, and never a file-name extension. + ``other`` is a well-formed content type outside the supported families; + ``unknown`` is a missing, malformed, or explicitly unknown one (see + :func:`format_for_content_type`). + """ + + PDF = "pdf" + JPEG = "jpeg" + PNG = "png" + TIFF = "tiff" + OTHER = "other" + UNKNOWN = "unknown" + + +FAMILY_FORMATS: Mapping[DocumentFamily, frozenset[FileFormat]] = MappingProxyType( + { + DocumentFamily.PDF: frozenset({FileFormat.PDF}), + DocumentFamily.SCANNED_IMAGE: frozenset( + {FileFormat.JPEG, FileFormat.PNG, FileFormat.TIFF} + ), + } +) + +# The one table every connector's bucketing SQL is generated from, so a file +# lands in the same bucket whichever warehouse indexed it. IANA-registered +# essences only: a common misspelling like `image/jpg` is not what a standard +# upload tool records, and guessing at aliases is how two connectors come to +# disagree. Lookup is on the essence, lower-cased, with parameters stripped. +CONTENT_TYPE_FORMATS: Mapping[str, FileFormat] = MappingProxyType( + { + "application/pdf": FileFormat.PDF, + "image/jpeg": FileFormat.JPEG, + "image/png": FileFormat.PNG, + "image/tiff": FileFormat.TIFF, + } +) + +# Content types that state the format is not known. S3 and several upload tools +# record these when they had nothing better, so counting them as `other` would +# report a format the storage layer explicitly said it did not know. +_UNKNOWN_CONTENT_TYPES = frozenset({"application/octet-stream", "binary/octet-stream"}) + +# RFC 6838 type/subtype, restricted-name characters only. +_MIME_ESSENCE = re.compile(r"[a-z0-9][a-z0-9!#$&^_.+-]*/[a-z0-9][a-z0-9!#$&^_.+-]*") + + +def format_for_content_type(content_type: str | None) -> FileFormat: + """The bucket a reported content type belongs in. + + This is the reference semantics a connector's warehouse-side bucketing + expression has to reproduce, and the thing its tests compare against. It is + not meant to run over file records in Python: bucketing happens inside the + warehouse, and only the counts per bucket come back. + """ + + if content_type is None: + return FileFormat.UNKNOWN + essence = content_type.split(";", 1)[0].strip().lower() + if not _MIME_ESSENCE.fullmatch(essence) or essence in _UNKNOWN_CONTENT_TYPES: + return FileFormat.UNKNOWN + return CONTENT_TYPE_FORMATS.get(essence, FileFormat.OTHER) + + +class ProcessingStatus(str, Enum): + """The fixed status vocabulary a stored processing result is counted under. + + A stored value outside it is counted as ``unknown`` and never returned, + because a status column is where customer strings and provider error bodies + live. + """ + + SUCCESS = "success" + FAILURE = "failure" + PARTIAL = "partial" + UNKNOWN = "unknown" + + +class ResultFormatName(str, Enum): + """The materialized result formats dex knows how to read. + + Three are a provider's documented native output, read through fixed paths; + ``mapped_columns`` is a pipeline that kept its diagnostics in plain columns + and discarded the parser payload. + """ + + BIGQUERY_DOCUMENT_AI = "bigquery_document_ai" + SNOWFLAKE_AI_PARSE_DOCUMENT = "snowflake_ai_parse_document" + DATABRICKS_AI_PARSE_DOCUMENT = "databricks_ai_parse_document" + MAPPED_COLUMNS = "mapped_columns" + + +class Diagnostic(str, Enum): + """The content diagnostics an assessment reports distributions for. + + Every one is a count computed inside the warehouse from a stored result. + Named with care: the envelope sanitizer refuses any key containing + ``token``, so a parser's token count can never be a diagnostic under that + name. + """ + + TEXT_CHARACTERS = "text_characters" + REPORTED_PAGES = "reported_pages" + REPRESENTED_PAGES = "represented_pages" + TABLES = "tables" + FORM_FIELDS = "form_fields" + PARAGRAPHS = "paragraphs" + + +# Each diagnostic's distribution is reported as fixed bins, each bin holding the +# files whose value is at least its edge and below the next one; the last bin is +# open. Fixed bins rather than percentiles because `SUM(CASE ...)` is the same SQL +# on every dialect and percentile functions are not. The first bin is exactly the +# observed zeros, which is what keeps "reported zero pages" apart from "no page +# count at all". +_COUNT_EDGES = (0, 1, 2, 6, 21, 101) +DIAGNOSTIC_BIN_EDGES: Mapping[Diagnostic, tuple[int, ...]] = MappingProxyType( + { + Diagnostic.TEXT_CHARACTERS: (0, 1, 100, 1_000, 10_000, 100_000), + Diagnostic.REPORTED_PAGES: _COUNT_EDGES, + Diagnostic.REPRESENTED_PAGES: _COUNT_EDGES, + Diagnostic.TABLES: _COUNT_EDGES, + Diagnostic.FORM_FIELDS: _COUNT_EDGES, + Diagnostic.PARAGRAPHS: _COUNT_EDGES, + } +) + + +class CollectionKind(str, Enum): + """What kind of warehouse resource indexes a collection.""" + + BIGQUERY_OBJECT_TABLE = "bigquery_object_table" + SNOWFLAKE_DIRECTORY_TABLE = "snowflake_directory_table" + DATABRICKS_VOLUME = "databricks_volume" + FILE_MANIFEST = "file_manifest" + + +class MetadataField(str, Enum): + """The per-file metadata a source can aggregate. + + Deliberately short. A file's path is its identity and is used only inside + the warehouse to match results; custom metadata key/value pairs and object + references are customer strings; and an object table's raw-bytes + pseudocolumn is the document itself. None of those is a field a source + declares, so none can be asked for. + """ + + SIZE = "size" + UPDATED = "updated" + CONTENT_TYPE = "content_type" + VERSION = "version" + + +class UnavailableReason(str, Enum): + """Why an aggregate field has no observed value. + + The reason is what makes an absent measurement distinguishable from a zero + one: an observed zero is a number, and everything that is not a number + carries one of these. + """ + + NOT_REPORTED = "not_reported" # the source's metadata does not carry it + NOT_BOUND = "not_bound" # the result binding maps no column for it + NOT_SUPPORTED_BY_FORMAT = "not_supported_by_format" + UNRECOGNIZED_SHAPE = "unrecognized_shape" # payload outside the declared shape + NO_EVIDENCE = "no_evidence" # nothing in the assessed set carried a value + NOT_ASSESSED = "not_assessed" # outside what this operation measures + + +class SamplingMethod(str, Enum): + """How an assessment chose its files. + + ``source_identity_hash`` orders the collection's files by a hash of their + source identity, breaks ties on the identity itself, and takes the first N, + all inside the warehouse. Reproducible for unchanged input, and chosen from + the source collection before any result is matched, so files with no result + stay in the sample. + """ + + SOURCE_IDENTITY_HASH = "source_identity_hash" + + +class Limitation(str, Enum): + """A fixed statement of what an aggregate does not establish. + + Codes rather than prose so that no limitation can quote the data it is + about; :data:`LIMITATION_TEXT` holds the engine's own wording. + """ + + DOCUMENT_PII_NOT_SCREENED = "document_pii_not_screened" + NATIVE_PROCESSING_UNAVAILABLE = "native_processing_unavailable" + SAMPLING_NOT_REPRESENTATIVE = "sampling_not_representative" + FORMAT_IS_REPORTED_METADATA = "format_is_reported_metadata" + COLLECTION_IS_REGISTERED_INDEX = "collection_is_registered_index" + PAGES_PARTIALLY_REPRESENTED = "pages_partially_represented" + VERSION_EVIDENCE_MISSING = "version_evidence_missing" + TIMESTAMP_NOT_VERSION_PROOF = "timestamp_not_version_proof" + + +LIMITATION_TEXT: Mapping[Limitation, str] = MappingProxyType( + { + Limitation.DOCUMENT_PII_NOT_SCREENED: ( + "document content was not screened for personal data, so the absence " + "of a finding says nothing about what the files contain" + ), + Limitation.NATIVE_PROCESSING_UNAVAILABLE: ( + "dex invoked no document processing; every figure describes results " + "that already existed in the warehouse" + ), + Limitation.SAMPLING_NOT_REPRESENTATIVE: ( + "the assessed files are a deterministic hash sample of source " + "identities, which bounds the work and is not a claim of statistical " + "representativeness" + ), + Limitation.FORMAT_IS_REPORTED_METADATA: ( + "formats come from the content type the storage layer reports, not " + "from inspecting file contents, and a file-name extension is never read" + ), + Limitation.COLLECTION_IS_REGISTERED_INDEX: ( + "counts describe what the registered collection indexes, which can " + "differ from everything in the underlying bucket or volume" + ), + Limitation.PAGES_PARTIALLY_REPRESENTED: ( + "some results cover only part of their document's pages, so those " + "documents were not assessed whole" + ), + Limitation.VERSION_EVIDENCE_MISSING: ( + "version evidence is missing on at least one side of the match, so " + "whether those results describe the current file is unknown" + ), + Limitation.TIMESTAMP_NOT_VERSION_PROOF: ( + "a processing timestamp orders attempts but does not prove which " + "version of a file was processed" + ), + } +) + +SHORTLIST_DEFAULT = 30 +SAMPLE_DEFAULT = 200 +SAMPLE_CEILING = 1000 + + +# --- Interfaces -------------------------------------------------------------- + + +class MeasureKind(str, Enum): + """What one alias of a file aggregate statement is allowed to return.""" + + COUNT = "count" + TIMESTAMP = "timestamp" + + +@runtime_checkable +class FileCollectionSource(Protocol): + """A connector that can aggregate a file collection's metadata. + + The declared attributes are static facts about the connector, read without + opening anything. ``document_families`` is what its metadata can bucket, and + ``metadata_fields`` is which per-file facts its collections carry, so a field + outside it is reported unavailable rather than defaulted. + + No member may read file content, refresh an external metadata cache, or + create a resource: a collection is read exactly as configured. + """ + + #: Stable connector name, the same one the warehouse adapter carries. + name: str + collection_kind: CollectionKind + metadata_fields: frozenset[MetadataField] + document_families: frozenset[DocumentFamily] + + def file_collection_inventory(self, collection: str) -> CollectionInventory: + """Aggregate one collection's metadata in a single budgeted statement. + + Admitted through the adapter's own cost gate like any other scan. A + count the collection's metadata does not carry comes back + :class:`~.results.Unavailable`, never zero. + """ + ... + + +@runtime_checkable +class DiscoveringFileSource(Protocol): + """A source that can also list the collections inside the source scope. + + Beside :class:`FileCollectionSource` rather than a member of it: a source + reading an explicitly bound manifest can aggregate it without being able to + discover anything, and declining discovery is an answer rather than a gap. + """ + + def list_file_collections(self) -> list[CollectionSummary]: + """Every collection in scope, from catalog metadata alone. + + Never scans a collection to manufacture a count: a count the catalog + does not carry is unavailable, because turning a listing into a scan per + collection turns a free command into a bill. + """ + ... + + +@runtime_checkable +class FileResultSource(FileCollectionSource, Protocol): + """A source that can also assess materialized processing results. + + The adapter owns the dialect and the connection; the orchestrator owns the + shape of the operation. So the adapter contributes the bounded source + selection and executes the one aggregate statement the orchestrator + assembles, and never returns a row. + """ + + def quote_identifier(self, name: str) -> str: + """``name`` quoted as one identifier in this connector's dialect.""" + ... + + def source_sample_sql( + self, + collection: str, + formats: frozenset[FileFormat], + sample_files: int, + ) -> str: + """A SELECT choosing at most ``sample_files`` files of ``formats``. + + Deterministic: ordered by a hash of the source identity with the + identity itself as the tie-breaker, as :attr:`SamplingMethod. + SOURCE_IDENTITY_HASH` describes. It yields exactly the columns + ``source_identity``, ``source_version`` (NULL where the collection + carries no version evidence), and ``source_format`` (a + :class:`FileFormat` value), and it is only ever embedded in a statement + that aggregates it. + """ + ... + + def run_file_aggregate( + self, sql: str, measures: Mapping[str, MeasureKind] + ) -> dict[str, int | datetime | None]: + """Execute one engine-assembled aggregate statement and return its row. + + Only the aliases named in ``measures`` come back, each checked against + its kind; a statement that yields more than one row, or a value of the + wrong kind, is refused rather than coerced. Charged to the adapter's + own cost gate. ``None`` is a SQL NULL, which the orchestrator turns into + an :class:`~.results.Unavailable` with the reason it knows and the + adapter does not. + """ + ... + + +@dataclass(frozen=True) +class RowDiagnostics: + """What a result format computes over one stored result row. + + Each value is a SQL expression over that row, or an + :class:`~.results.Unavailable` where the format does not support the + measurement. ``status`` must evaluate to a :class:`ProcessingStatus` value, + and ``payload_valid`` to a boolean saying whether the stored payload has the + shape the format declares. Every :class:`Diagnostic` has an entry, so an + unsupported one is stated rather than silently absent. + """ + + status: str | Unavailable + payload_valid: str | Unavailable + diagnostics: Mapping[Diagnostic, str | Unavailable] + + def __post_init__(self) -> None: + missing = [d.value for d in Diagnostic if d not in self.diagnostics] + if missing: + raise ValueError( + "a result format must state every diagnostic, as an expression or " + f"as unavailable; missing: {', '.join(missing)}" + ) + object.__setattr__( + self, "diagnostics", MappingProxyType(dict(self.diagnostics)) + ) + + +@runtime_checkable +class ResultFormat(Protocol): + """Interpretation of one materialized result format. + + It generates SQL expressions and nothing else. It never executes a + statement, and in particular never calls a document-processing function: + a result format reads what a previous pipeline stored. + """ + + name: ResultFormatName + #: The connectors whose dialect this format's expressions are written in. + connectors: frozenset[str] + document_families: frozenset[DocumentFamily] + + def row_diagnostics( + self, binding: ResultBinding, quote: Callable[[str], str] + ) -> RowDiagnostics: + """The per-row expressions for ``binding``'s result table. + + ``quote`` is the source's identifier quoting, passed in so a format that + spans dialects never has to know which one it is writing for. + """ + ... + + +# The registered result formats. Empty until a format adapter is implemented, so +# today every connector reports result assessment unavailable by name. +RESULT_FORMATS: tuple[ResultFormat, ...] = () + + +# --- Requests ---------------------------------------------------------------- + + +@dataclass(frozen=True) +class ResultBinding: + """A collection bound to the materialized table holding its results. + + The engine-side, already-resolved form. Project configuration is where a + binding is authored and validated, including that every relation it names is + inside the source scope and that every column is a plain identifier; this is + what that validation produces. It carries identifiers only, never an + expression, which is what keeps a binding from being a way to run SQL. + + A native format reads ``payload_column`` (and ``status_column`` where the + provider stores one separately). ``mapped_columns`` maps diagnostics onto + plain columns for the ``mapped_columns`` format, with ``status_column`` as + the status. ``version_column`` and ``processed_at_column`` are optional + evidence: without the first, correspondence to the current file is unknown; + without the second, duplicate results cannot be resolved to a latest one. + """ + + collection: str + result_table: str + identity_column: str + result_format: ResultFormatName + payload_column: str | None = None + status_column: str | None = None + mapped_columns: Mapping[Diagnostic, str] = field(default_factory=dict) + version_column: str | None = None + processed_at_column: str | None = None + + def __post_init__(self) -> None: + if self.result_format is ResultFormatName.MAPPED_COLUMNS: + if self.payload_column is not None: + raise RequestError( + "a mapped_columns binding maps diagnostics onto columns and " + "reads no parser payload; drop payload_column" + ) + if not self.mapped_columns and self.status_column is None: + raise RequestError( + "a mapped_columns binding needs a status_column or at least " + "one mapped diagnostic column" + ) + else: + if self.payload_column is None: + raise RequestError( + f"a {self.result_format.value} binding reads the provider's " + "stored payload; name its payload_column" + ) + if self.mapped_columns: + raise RequestError( + f"a {self.result_format.value} binding reads fixed payload " + "paths; mapped_columns applies only to the mapped_columns format" + ) + object.__setattr__( + self, "mapped_columns", MappingProxyType(dict(self.mapped_columns)) + ) + + +@dataclass(frozen=True) +class InventoryRequest: + """List the collections in scope, or aggregate one. + + ``limit`` and ``show_all`` shape the shortlist of a listing; neither ever + triggers a scan of a collection. + """ + + collection: str | None = None + limit: int = SHORTLIST_DEFAULT + show_all: bool = False + + def __post_init__(self) -> None: + if isinstance(self.limit, bool) or not isinstance(self.limit, int): + raise RequestError(f"limit must be a whole number, not {self.limit!r}") + if self.limit < 1: + raise RequestError(f"limit must be at least 1, not {self.limit}") + + +@dataclass(frozen=True) +class ProfileRequest: + """Assess a collection's materialized results over a deterministic sample. + + ``families`` accepts the family values as strings too, since that is how a + command line delivers them, and defaults to every family. ``sample_files`` + bounds the assessment set, not the warehouse's scan, which the statement's + own estimate prices at admission. + """ + + collection: str + binding: ResultBinding + families: frozenset[DocumentFamily] = frozenset(DocumentFamily) + sample_files: int = SAMPLE_DEFAULT + + def __post_init__(self) -> None: + families: set[DocumentFamily] = set() + for value in self.families: + try: + families.add(DocumentFamily(value)) + except ValueError: + allowed = ", ".join(f.value for f in DocumentFamily) + raise RequestError( + f"'{value}' is not a document family dex assesses; use one of " + f"{allowed}" + ) from None + if not families: + raise RequestError("name at least one document family to assess") + object.__setattr__(self, "families", frozenset(families)) + + sample = self.sample_files + if isinstance(sample, bool) or not isinstance(sample, int): + raise RequestError(f"sample_files must be a whole number, not {sample!r}") + if not 1 <= sample <= SAMPLE_CEILING: + raise RequestError( + f"sample_files must be between 1 and {SAMPLE_CEILING}, not {sample}; " + "there is no full-collection content assessment" + ) + if self.binding.collection != self.collection: + raise RequestError( + f"the result binding is for '{self.binding.collection}', not " + f"'{self.collection}'" + ) + + @property + def formats(self) -> frozenset[FileFormat]: + """The format buckets the selected families cover.""" + + return frozenset().union(*(FAMILY_FORMATS[f] for f in self.families)) + + +# --- Capabilities ------------------------------------------------------------ + + +class CapabilityLimitation(str, Enum): + """Why a file capability is unavailable on this connection.""" + + NO_FILE_SOURCE = "no_file_source" + NO_COLLECTION_DISCOVERY = "no_collection_discovery" + NO_RESULT_SOURCE = "no_result_source" + NO_RESULT_FORMAT = "no_result_format" + + +_CAPABILITY_LIMITATION_TEXT: Mapping[CapabilityLimitation, str] = MappingProxyType( + { + CapabilityLimitation.NO_FILE_SOURCE: ( + "this connector implements no file-exploration source; dex does not " + "read object storage on its own and does not switch to another connector" + ), + CapabilityLimitation.NO_COLLECTION_DISCOVERY: ( + "this connector aggregates a collection it is given but cannot list " + "collections, so name one explicitly" + ), + CapabilityLimitation.NO_RESULT_SOURCE: ( + "this connector can aggregate collection metadata but cannot assess " + "materialized processing results" + ), + CapabilityLimitation.NO_RESULT_FORMAT: ( + "no supported result format is written for this connector's dialect" + ), + } +) + + +class Capability(BaseModel): + """One file capability: available, or unavailable with its named reason.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + available: bool + limitation: CapabilityLimitation | None = None + + @model_validator(mode="after") + def _limitation_explains_absence(self) -> Capability: + if self.available and self.limitation is not None: + raise ValueError("an available capability carries no limitation") + if not self.available and self.limitation is None: + raise ValueError("an unavailable capability must name its limitation") + return self + + @computed_field # type: ignore[prop-decorator] + @property + def explanation(self) -> str | None: + if self.limitation is None: + return None + return _CAPABILITY_LIMITATION_TEXT[self.limitation] + + +class NativeProcessing(BaseModel): + """Whether dex may invoke document processing: never, and why. + + There is no field to set. ``available`` and ``reason`` are computed + constants, so no construction, copy, deserialization, or confirmation flag + can report it available. Opening it requires a provider-enforced hard + monetary cap that covers the exact operation, holds against concurrent and + in-flight work, and that dex can verify; estimates, page limits, delayed + quotas, and user confirmation do not substitute for one. Changing that is a + code change reviewed on its own, never a configuration. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + @computed_field # type: ignore[prop-decorator] + @property + def available(self) -> bool: + return False + + @computed_field # type: ignore[prop-decorator] + @property + def reason(self) -> str: + return "no_verified_hard_spend_cap" + + @computed_field # type: ignore[prop-decorator] + @property + def explanation(self) -> str: + return ( + "new document processing needs a provider-enforced hard spend cap " + "that dex can verify, and no supported processing path offers one; " + "dex assesses results that already exist and never invokes a " + "document processor" + ) + + +class FileCapabilities(BaseModel): + """What file exploration this connection supports, derived structurally.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + collection_discovery: Capability + metadata_aggregation: Capability + result_assessment: Capability + document_families: tuple[DocumentFamily, ...] + result_formats: tuple[ResultFormatName, ...] + native_processing: NativeProcessing = NativeProcessing() + + def payload(self) -> dict[str, Any]: + """The JSON shape a capability report carries.""" + + return self.model_dump(mode="json") + + +def _in_declared_order(values, enum: type[Enum]) -> tuple: + chosen = set(values) + return tuple(member for member in enum if member in chosen) + + +def file_capabilities( + adapter: object, result_formats: tuple[ResultFormat, ...] = RESULT_FORMATS +) -> FileCapabilities: + """What ``adapter`` can do with files, read off the protocols it satisfies. + + Nothing is probed and nothing is claimed: a connector that implements no + file source gets every capability unavailable with the same named reason, + and this never looks for another connector to answer instead. Result + formats count only when they are written for this connector's dialect. + """ + + if not isinstance(adapter, FileCollectionSource): + absent = Capability( + available=False, limitation=CapabilityLimitation.NO_FILE_SOURCE + ) + return FileCapabilities( + collection_discovery=absent, + metadata_aggregation=absent, + result_assessment=absent, + document_families=(), + result_formats=(), + ) + + available = Capability(available=True) + discovery = ( + available + if isinstance(adapter, DiscoveringFileSource) + else Capability( + available=False, limitation=CapabilityLimitation.NO_COLLECTION_DISCOVERY + ) + ) + + formats: tuple[ResultFormatName, ...] = () + if not isinstance(adapter, FileResultSource): + assessment = Capability( + available=False, limitation=CapabilityLimitation.NO_RESULT_SOURCE + ) + else: + formats = _in_declared_order( + (f.name for f in result_formats if adapter.name in f.connectors), + ResultFormatName, + ) + assessment = ( + available + if formats + else Capability( + available=False, limitation=CapabilityLimitation.NO_RESULT_FORMAT + ) + ) + + return FileCapabilities( + collection_discovery=discovery, + metadata_aggregation=available, + result_assessment=assessment, + document_families=_in_declared_order(adapter.document_families, DocumentFamily), + result_formats=formats, + ) diff --git a/packages/dex-core/src/exmergo_dex_core/files/results.py b/packages/dex-core/src/exmergo_dex_core/files/results.py new file mode 100644 index 00000000..192d471e --- /dev/null +++ b/packages/dex-core/src/exmergo_dex_core/files/results.py @@ -0,0 +1,488 @@ +"""The aggregates a file operation returns, and nothing else. + +These are the only types that cross from a file source into the command layer, +and they are built so that they *cannot* carry document content, rather than so +that they are merely not given any. Three properties do that, and +``tests/files/test_file_results.py`` walks every model to hold them: + +- **Every leaf is a number, a fixed category, a timestamp, or a relation name.** + Counts are strict non-negative integers (a string of digits is refused, not + coerced), categories are the enums in :mod:`.contract`, and the one string + type, :data:`RelationName`, refuses anything shaped like a path or a URL. There + is no free-text field anywhere, so there is nowhere for an excerpt, a file + name, or a provider's error body to go. +- **Unknown keys are refused.** Every model forbids extra fields, so a ``uri`` or + a ``text`` cannot ride along on an otherwise valid aggregate. +- **Absent and zero never share a spelling.** No field admits ``None``. An + observed zero is ``0``; a measurement the evidence does not support is an + :class:`Unavailable` carrying the reason. A caller can always tell "no file had + a table" from "tables were never counted". + +The models also refuse internally inconsistent reports: the coverage groups must +partition the sample, so files with no result can never drop out of the +denominator, and the mandatory limitations must be present, so an aggregate can +never be read as saying more than it measured. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Annotated, Any + +from pydantic import ( + AwareDatetime, + BaseModel, + ConfigDict, + Field, + Strict, + StringConstraints, + computed_field, + model_validator, +) + +from .contract import ( + DIAGNOSTIC_BIN_EDGES, + SAMPLE_CEILING, + CollectionKind, + Diagnostic, + DocumentFamily, + Limitation, + MetadataField, + ResultFormatName, + SamplingMethod, + UnavailableReason, +) + +__all__ = [ + "RELATION_NAME_PATTERN", + "Bin", + "CollectionInventory", + "CollectionScope", + "CollectionSummary", + "Count", + "Coverage", + "Currency", + "DiagnosticDistributions", + "Distribution", + "FileAggregate", + "FileProfile", + "FormatCounts", + "Instant", + "NonNegative", + "Processing", + "Ratio", + "RelationName", + "StatusCounts", + "Unavailable", +] + + +NonNegative = Annotated[int, Strict(), Field(ge=0)] + +# Dotted identifier parts: letters, digits, underscore, `$`, and the hyphen a +# BigQuery project id carries. What it exists to refuse is everything a path or a +# URL needs (`/`, `:`, `?`, `=`, `&`, `%`, whitespace, quotes), so a `gs://` URI or +# a signed URL can never be stored as a collection name. It cannot tell a bare +# file name like `scan.pdf` from a two-part relation, and does not try: the +# guarantee against file names is that no field is meant to hold one, and this +# pattern is the second line behind it. +RELATION_NAME_PATTERN = r"^[\w$-]+(?:\.[\w$-]+)*$" +RelationName = Annotated[ + str, StringConstraints(pattern=RELATION_NAME_PATTERN, max_length=1024) +] + + +class FileAggregate(BaseModel): + """Base for every file aggregate: immutable, and closed to unknown keys.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + def payload(self) -> dict[str, Any]: + """The JSON shape this aggregate reports as.""" + + return self.model_dump(mode="json") + + +class Unavailable(FileAggregate): + """A measurement with no observed value, and why.""" + + reason: UnavailableReason + + +Count = NonNegative | Unavailable +Instant = AwareDatetime | Unavailable + + +def _limitations_present( + limitations: tuple[Limitation, ...], required: frozenset[Limitation], what: str +) -> None: + if len(set(limitations)) != len(limitations): + raise ValueError(f"{what} names a limitation twice") + missing = sorted(lim.value for lim in required - set(limitations)) + if missing: + raise ValueError(f"{what} must state its limitations: {', '.join(missing)}") + + +class Ratio(FileAggregate): + """A rate that always carries what it is a rate of.""" + + numerator: NonNegative + denominator: NonNegative + + @model_validator(mode="before") + @classmethod + def _accept_own_fraction(cls, data: Any) -> Any: + """Read back a ratio this model serialized, which carries ``fraction``. + + The fraction is derived, so a stored one is accepted only when it is the + value the two counts produce; anything else is refused rather than + silently recomputed, because a disagreeing fraction means the record was + edited. + """ + + if not isinstance(data, dict) or "fraction" not in data: + return data + data = dict(data) + stored = data.pop("fraction") + numerator, denominator = data.get("numerator"), data.get("denominator") + if isinstance(numerator, int) and isinstance(denominator, int): + derived = numerator / denominator if denominator else None + if stored != derived: + raise ValueError( + f"a stored fraction of {stored} does not follow from " + f"{numerator} over {denominator}" + ) + return data + + @model_validator(mode="after") + def _within_denominator(self) -> Ratio: + if self.numerator > self.denominator: + raise ValueError( + f"a ratio's numerator ({self.numerator}) cannot exceed its " + f"denominator ({self.denominator})" + ) + return self + + @computed_field # type: ignore[prop-decorator] + @property + def fraction(self) -> float | None: + """``None`` only over an empty denominator, where no rate exists.""" + + return self.numerator / self.denominator if self.denominator else None + + +class FormatCounts(FileAggregate): + """Files per format bucket. Every file lands in exactly one, so the buckets + sum to the file count, and a file whose metadata carries no content type is + counted under ``unknown`` rather than dropped.""" + + pdf: NonNegative + jpeg: NonNegative + png: NonNegative + tiff: NonNegative + other: NonNegative + unknown: NonNegative + + @property + def total(self) -> int: + return self.pdf + self.jpeg + self.png + self.tiff + self.other + self.unknown + + +class CollectionSummary(FileAggregate): + """One collection as discovery sees it, from catalog metadata only. + + ``file_count`` is usually unavailable here, because discovery never scans a + collection to count it. + """ + + collection: RelationName + kind: CollectionKind + file_count: Count + metadata_refreshed_at: Instant + + +_INVENTORY_LIMITATIONS = frozenset( + { + Limitation.FORMAT_IS_REPORTED_METADATA, + Limitation.COLLECTION_IS_REGISTERED_INDEX, + } +) + + +class CollectionInventory(FileAggregate): + """One collection's metadata, aggregated inside the warehouse. + + ``file_count`` is always observed: an inventory is a count. Everything else + depends on what the collection's metadata carries, so each is a + :data:`Count` or an :data:`Instant`. ``missing_size`` counts files whose size + is unknown, and ``known_bytes`` sums only the rest, which is why the two are + reported side by side. + """ + + collection: RelationName + kind: CollectionKind + observed_at: AwareDatetime + file_count: NonNegative + known_bytes: Count + missing_size: Count + zero_byte: Count + formats: FormatCounts + earliest_update: Instant + latest_update: Instant + metadata_refreshed_at: Instant + limitations: tuple[Limitation, ...] + + @model_validator(mode="after") + def _consistent(self) -> CollectionInventory: + _limitations_present( + self.limitations, _INVENTORY_LIMITATIONS, "a collection inventory" + ) + if self.formats.total != self.file_count: + raise ValueError( + f"format buckets sum to {self.formats.total}, not the " + f"{self.file_count} files counted" + ) + missing = self.missing_size if isinstance(self.missing_size, int) else 0 + if missing > self.file_count: + raise ValueError("more files are missing a size than were counted") + if ( + isinstance(self.zero_byte, int) + and self.zero_byte > self.file_count - missing + ): + raise ValueError("more files are zero bytes than have a known size") + if ( + isinstance(self.earliest_update, datetime) + and isinstance(self.latest_update, datetime) + and self.earliest_update > self.latest_update + ): + raise ValueError("the earliest update is later than the latest one") + return self + + +class Bin(FileAggregate): + """Files whose value is at least ``at_least`` and below the next bin's edge. + + The last bin of a distribution is open-ended. + """ + + at_least: NonNegative + files: NonNegative + + +class Distribution(FileAggregate): + """One diagnostic over the assessed files. + + ``assessed`` is partitioned three ways: a usable value, no value at all + (``files_absent``), or a value that cannot be one (``files_invalid``: a + negative page count, a type the format does not declare). The first bin is + the observed zeros, so zero and absent stay apart here too. + """ + + assessed: NonNegative + files_with_value: NonNegative + files_absent: NonNegative + files_invalid: NonNegative + minimum: Count + maximum: Count + bins: tuple[Bin, ...] + + @model_validator(mode="after") + def _consistent(self) -> Distribution: + total = self.files_with_value + self.files_absent + self.files_invalid + if total != self.assessed: + raise ValueError( + f"with-value, absent, and invalid files sum to {total}, not the " + f"{self.assessed} assessed" + ) + edges = [b.at_least for b in self.bins] + if not edges or edges[0] != 0 or edges != sorted(set(edges)): + raise ValueError("bins must start at 0 and rise strictly") + if sum(b.files for b in self.bins) != self.files_with_value: + raise ValueError("bins must hold exactly the files with a value") + if self.files_with_value == 0: + if isinstance(self.minimum, int) or isinstance(self.maximum, int): + raise ValueError( + "no file had a value, so there is no minimum or maximum" + ) + elif not (isinstance(self.minimum, int) and isinstance(self.maximum, int)): + raise ValueError( + "files had values, so the minimum and maximum are observed" + ) + elif self.minimum > self.maximum: + raise ValueError("the minimum exceeds the maximum") + return self + + +class DiagnosticDistributions(FileAggregate): + """One entry per :class:`~.contract.Diagnostic`, each a distribution or the + reason there is none.""" + + text_characters: Distribution | Unavailable + reported_pages: Distribution | Unavailable + represented_pages: Distribution | Unavailable + tables: Distribution | Unavailable + form_fields: Distribution | Unavailable + paragraphs: Distribution | Unavailable + + def by_diagnostic(self) -> dict[Diagnostic, Distribution | Unavailable]: + return {d: getattr(self, d.value) for d in Diagnostic} + + @model_validator(mode="after") + def _fixed_bins(self) -> DiagnosticDistributions: + for diagnostic, entry in self.by_diagnostic().items(): + if not isinstance(entry, Distribution): + continue + edges = tuple(b.at_least for b in entry.bins) + if edges != DIAGNOSTIC_BIN_EDGES[diagnostic]: + raise ValueError( + f"{diagnostic.value} must use its fixed bin edges " + f"{DIAGNOSTIC_BIN_EDGES[diagnostic]}" + ) + return self + + +class StatusCounts(FileAggregate): + """Matched files per processing status; anything else a provider stored is + counted under ``unknown`` and never returned.""" + + success: NonNegative + failure: NonNegative + partial: NonNegative + unknown: NonNegative + + @property + def total(self) -> int: + return self.success + self.failure + self.partial + self.unknown + + +class CollectionScope(FileAggregate): + """What was assessed: which collection, against which results, for which + families, with which metadata available.""" + + collection: RelationName + kind: CollectionKind + result_table: RelationName + result_format: ResultFormatName + families: tuple[DocumentFamily, ...] = Field(min_length=1) + metadata_fields: tuple[MetadataField, ...] + + +class Coverage(FileAggregate): + """How much of the sample has a result at all. + + ``eligible`` is the collection's files in the selected formats and + ``sampled`` the ones chosen from them. The sample splits three ways, and the + three always add back up to it: one resolvable result (``matched``), no + result (``missing_result``), or duplicates that could not be resolved to one + (``ambiguous``), which are excluded from every content distribution rather + than arbitrarily picked. + """ + + requested: Annotated[int, Strict(), Field(ge=1, le=SAMPLE_CEILING)] + eligible: NonNegative + sampled: NonNegative + sampling_method: SamplingMethod + matched: Ratio + missing_result: Ratio + ambiguous: Ratio + + @model_validator(mode="after") + def _partitions_the_sample(self) -> Coverage: + if self.sampled != min(self.requested, self.eligible): + raise ValueError( + "a deterministic sample takes every eligible file up to the " + "requested size, so sampled must equal the smaller of the two" + ) + parts = (self.matched, self.missing_result, self.ambiguous) + if any(p.denominator != self.sampled for p in parts): + raise ValueError("every coverage rate is over the sampled files") + if sum(p.numerator for p in parts) != self.sampled: + raise ValueError( + "matched, missing, and ambiguous files must add up to the sample; " + "a file with no result stays in the denominator" + ) + return self + + +class Currency(FileAggregate): + """Whether each matched result describes the file as it is now. + + Compared only where both sides carry compatible version evidence; everything + else is ``version_unknown``, never current. + """ + + version_matched: Ratio + version_mismatched: Ratio + version_unknown: Ratio + + @model_validator(mode="after") + def _partitions_matched(self) -> Currency: + parts = (self.version_matched, self.version_mismatched, self.version_unknown) + denominators = {p.denominator for p in parts} + if len(denominators) != 1: + raise ValueError("every currency rate is over the same matched files") + if sum(p.numerator for p in parts) != denominators.pop(): + raise ValueError("matched, mismatched, and unknown versions must add up") + return self + + +class Processing(FileAggregate): + """What the matched results report about their own processing. + + ``payload_valid`` is whether a stored payload has the shape its format + declares, which is separate from coverage: a file can have a result whose + payload is unusable. + """ + + statuses: StatusCounts + payload_valid: Ratio | Unavailable + partial_page_files: Count + diagnostics: DiagnosticDistributions + + +_PROFILE_LIMITATIONS = frozenset( + { + Limitation.DOCUMENT_PII_NOT_SCREENED, + Limitation.NATIVE_PROCESSING_UNAVAILABLE, + Limitation.SAMPLING_NOT_REPRESENTATIVE, + } +) + + +class FileProfile(FileAggregate): + """An assessment of a collection's existing processing results. + + Five groups: what was assessed, how much of it has results, whether those + results are current, what they report, and what the assessment does not + establish. There is deliberately no overall readiness score: successful + parsing, long text, or detected tables do not make an extraction correct. + """ + + collection: CollectionScope + observed_at: AwareDatetime + coverage: Coverage + currency: Currency + processing: Processing + limitations: tuple[Limitation, ...] + + @model_validator(mode="after") + def _consistent(self) -> FileProfile: + _limitations_present(self.limitations, _PROFILE_LIMITATIONS, "a file profile") + matched = self.coverage.matched.numerator + if self.currency.version_matched.denominator != matched: + raise ValueError("currency is measured over the matched files") + if self.processing.statuses.total != matched: + raise ValueError("processing statuses are counted over the matched files") + valid = self.processing.payload_valid + if isinstance(valid, Ratio) and valid.denominator != matched: + raise ValueError("payload validity is measured over the matched files") + partial = self.processing.partial_page_files + if isinstance(partial, int) and partial > matched: + raise ValueError("more files are partially represented than matched") + for diagnostic, entry in self.processing.diagnostics.by_diagnostic().items(): + if isinstance(entry, Distribution) and entry.assessed != matched: + raise ValueError( + f"{diagnostic.value} is distributed over the matched files, " + "which excludes ambiguous duplicates" + ) + return self diff --git a/packages/dex-core/tests/files/test_file_contract.py b/packages/dex-core/tests/files/test_file_contract.py new file mode 100644 index 00000000..8c956693 --- /dev/null +++ b/packages/dex-core/tests/files/test_file_contract.py @@ -0,0 +1,469 @@ +"""The file-exploration contract: fixed vocabularies, request refusals, and a +capability model read off the protocols a connector satisfies rather than off +anything it claims.""" + +from __future__ import annotations + +import pytest + +from exmergo_dex_core import envelope as env +from exmergo_dex_core.adapters import _adapter_class +from exmergo_dex_core.adapters.base import Adapter +from exmergo_dex_core.errors import RequestError +from exmergo_dex_core.files.contract import ( + DIAGNOSTIC_BIN_EDGES, + FAMILY_FORMATS, + RESULT_FORMATS, + SAMPLE_CEILING, + Capability, + CapabilityLimitation, + CollectionKind, + Diagnostic, + DiscoveringFileSource, + DocumentFamily, + FileCollectionSource, + FileFormat, + FileResultSource, + InventoryRequest, + Limitation, + MeasureKind, + MetadataField, + NativeProcessing, + ProcessingStatus, + ProfileRequest, + ResultBinding, + ResultFormatName, + RowDiagnostics, + UnavailableReason, + file_capabilities, + format_for_content_type, +) + +SHIPPED_CONNECTORS = [ + "duckdb", + "bigquery", + "snowflake", + "databricks", + "postgres", + "redshift", + "clickhouse", +] + +# --- vocabulary --------------------------------------------------------------- + + +def test_scanned_image_is_exactly_the_three_document_image_formats(): + assert FAMILY_FORMATS[DocumentFamily.SCANNED_IMAGE] == { + FileFormat.JPEG, + FileFormat.PNG, + FileFormat.TIFF, + } + assert FAMILY_FORMATS[DocumentFamily.PDF] == {FileFormat.PDF} + + +def test_every_format_belongs_to_at_most_one_family_and_other_to_none(): + claimed = [fmt for formats in FAMILY_FORMATS.values() for fmt in formats] + assert len(claimed) == len(set(claimed)) + assert FileFormat.OTHER not in claimed + assert FileFormat.UNKNOWN not in claimed + + +@pytest.mark.parametrize( + ("content_type", "expected"), + [ + ("application/pdf", FileFormat.PDF), + ("Application/PDF; version=1.7", FileFormat.PDF), + (" image/jpeg ", FileFormat.JPEG), + ("image/png", FileFormat.PNG), + ("image/tiff", FileFormat.TIFF), + # Well formed, outside the supported families. + ("text/plain", FileFormat.OTHER), + ( + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + FileFormat.OTHER, + ), + # A common misspelling is not what an upload tool records, and guessing + # at aliases is how two connectors come to disagree. + ("image/jpg", FileFormat.OTHER), + # The storage layer saying it does not know. + ("application/octet-stream", FileFormat.UNKNOWN), + ("binary/octet-stream", FileFormat.UNKNOWN), + # Missing or malformed. + (None, FileFormat.UNKNOWN), + ("", FileFormat.UNKNOWN), + ("pdf", FileFormat.UNKNOWN), + ("invoice.pdf", FileFormat.UNKNOWN), + ], +) +def test_a_reported_content_type_lands_in_one_fixed_bucket(content_type, expected): + assert format_for_content_type(content_type) is expected + + +def test_a_file_name_extension_never_decides_the_format(): + """The bucket is read off the reported content type alone, so a PNG that was + uploaded under a `.pdf` name counts as a PNG.""" + + assert format_for_content_type("image/png") is FileFormat.PNG + assert format_for_content_type("scan.pdf") is FileFormat.UNKNOWN + + +def test_every_diagnostic_has_fixed_bin_edges_starting_at_zero(): + for diagnostic in Diagnostic: + edges = DIAGNOSTIC_BIN_EDGES[diagnostic] + assert edges[0] == 0 + assert list(edges) == sorted(set(edges)) + # The first bin is exactly the observed zeros. + assert all(DIAGNOSTIC_BIN_EDGES[d][1] == 1 for d in Diagnostic) + + +def test_no_vocabulary_value_trips_the_envelope_key_screens(): + """Diagnostics become payload keys, and the sanitizer refuses any key that + reads like a secret or like row data. A diagnostic named after a parser's + `tokens` would take every profile down on the way out.""" + + screens = (*env._SECRET_KEY_PATTERNS, *env._RAW_ROW_KEY_PATTERNS) + for enum in (Diagnostic, FileFormat, ProcessingStatus, MetadataField): + for member in enum: + assert not any(p in member.value for p in screens), member + + +# --- requests ----------------------------------------------------------------- + + +def _binding(**overrides) -> ResultBinding: + fields = { + "collection": "proj.docs.files_obj", + "result_table": "proj.docs.parsed", + "identity_column": "uri", + "result_format": ResultFormatName.BIGQUERY_DOCUMENT_AI, + "payload_column": "ml_process_document_result", + "status_column": "ml_process_document_status", + } + fields.update(overrides) + return ResultBinding(**fields) + + +def test_a_profile_request_defaults_to_every_family_and_two_hundred_files(): + request = ProfileRequest(collection="proj.docs.files_obj", binding=_binding()) + assert request.families == set(DocumentFamily) + assert request.sample_files == 200 + assert request.formats == { + FileFormat.PDF, + FileFormat.JPEG, + FileFormat.PNG, + FileFormat.TIFF, + } + + +def test_a_profile_request_takes_family_names_the_way_a_command_line_sends_them(): + request = ProfileRequest( + collection="proj.docs.files_obj", + binding=_binding(), + families=frozenset({"scanned_image"}), + ) + assert request.families == {DocumentFamily.SCANNED_IMAGE} + assert request.formats == FAMILY_FORMATS[DocumentFamily.SCANNED_IMAGE] + + +@pytest.mark.parametrize("sample", [0, -1, SAMPLE_CEILING + 1, True, "200", 2.5]) +def test_a_profile_request_refuses_a_sample_outside_one_to_the_ceiling(sample): + with pytest.raises(RequestError, match="sample_files"): + ProfileRequest( + collection="proj.docs.files_obj", binding=_binding(), sample_files=sample + ) + + +def test_the_ceiling_itself_is_accepted_and_there_is_no_full_mode(): + request = ProfileRequest( + collection="proj.docs.files_obj", + binding=_binding(), + sample_files=SAMPLE_CEILING, + ) + assert request.sample_files == 1000 + + +def test_an_unknown_family_is_refused_by_name_with_the_ones_that_exist(): + with pytest.raises(RequestError) as refused: + ProfileRequest( + collection="proj.docs.files_obj", + binding=_binding(), + families=frozenset({"docx"}), + ) + assert "'docx'" in str(refused.value) + assert "pdf, scanned_image" in str(refused.value) + + +def test_a_profile_request_needs_at_least_one_family(): + with pytest.raises(RequestError, match="at least one"): + ProfileRequest( + collection="proj.docs.files_obj", binding=_binding(), families=frozenset() + ) + + +def test_a_binding_for_another_collection_is_refused(): + with pytest.raises(RequestError, match=r"proj\.docs\.files_obj"): + ProfileRequest(collection="proj.docs.other_obj", binding=_binding()) + + +@pytest.mark.parametrize("limit", [0, -5, True, "30"]) +def test_an_inventory_limit_must_be_a_positive_whole_number(limit): + with pytest.raises(RequestError, match="limit"): + InventoryRequest(limit=limit) + + +def test_an_inventory_request_defaults_to_a_shortlist_of_thirty(): + request = InventoryRequest() + assert (request.collection, request.limit, request.show_all) == (None, 30, False) + + +def test_a_native_binding_needs_its_payload_column(): + with pytest.raises(RequestError, match="payload_column"): + _binding(payload_column=None) + + +def test_a_native_binding_refuses_mapped_columns(): + with pytest.raises(RequestError, match="mapped_columns"): + _binding(mapped_columns={Diagnostic.REPORTED_PAGES: "page_count"}) + + +def test_a_mapped_binding_reads_no_payload(): + with pytest.raises(RequestError, match="payload_column"): + _binding(result_format=ResultFormatName.MAPPED_COLUMNS) + + +def test_a_mapped_binding_needs_something_to_read(): + with pytest.raises(RequestError, match="status_column"): + _binding( + result_format=ResultFormatName.MAPPED_COLUMNS, + payload_column=None, + status_column=None, + ) + + +def test_a_bindings_column_mapping_cannot_be_changed_after_it_is_resolved(): + mapping = {Diagnostic.REPORTED_PAGES: "page_count"} + binding = _binding( + result_format=ResultFormatName.MAPPED_COLUMNS, + payload_column=None, + mapped_columns=mapping, + ) + mapping[Diagnostic.TABLES] = "table_count" + assert Diagnostic.TABLES not in binding.mapped_columns + with pytest.raises(TypeError): + binding.mapped_columns[Diagnostic.TABLES] = "table_count" + + +def test_a_result_format_has_to_state_every_diagnostic(): + from exmergo_dex_core.files.results import Unavailable + + unsupported = Unavailable(reason=UnavailableReason.NOT_SUPPORTED_BY_FORMAT) + with pytest.raises(ValueError, match="paragraphs"): + RowDiagnostics( + status="'success'", + payload_valid="TRUE", + diagnostics={ + d: unsupported for d in Diagnostic if d != Diagnostic.PARAGRAPHS + }, + ) + + +# --- compatibility -------------------------------------------------------------- + + +@pytest.mark.parametrize("connector", SHIPPED_CONNECTORS) +def test_no_shipped_connector_implements_a_file_source(connector): + """Existing connectors stay compatible by implementing nothing, and each + reports the absence as a named limitation rather than an empty answer. + Built without ``__init__`` so no connection is opened: the check is + structural and needs none.""" + + adapter = object.__new__(_adapter_class(connector)) + assert not isinstance(adapter, FileCollectionSource) + + capabilities = file_capabilities(adapter) + for capability in ( + capabilities.collection_discovery, + capabilities.metadata_aggregation, + capabilities.result_assessment, + ): + assert capability.available is False + assert capability.limitation is CapabilityLimitation.NO_FILE_SOURCE + assert capabilities.document_families == () + assert capabilities.result_formats == () + assert capabilities.native_processing.available is False + + +def test_the_warehouse_adapter_protocol_gained_no_file_member(): + """A new member on the runtime-checkable ``Adapter`` would demote every + host-supplied adapter that has not grown it.""" + + members = set(vars(Adapter)) | set(Adapter.__annotations__) + file_members = { + "file_collection_inventory", + "list_file_collections", + "source_sample_sql", + "run_file_aggregate", + "collection_kind", + "metadata_fields", + "document_families", + } + assert not members & file_members + + +# --- capabilities ----------------------------------------------------------------- + + +class _AggregatingSource: + """Tier 1 only. Every member refuses to run, because deriving capabilities + must never probe a connection.""" + + name = "fakewh" + collection_kind = CollectionKind.FILE_MANIFEST + metadata_fields = frozenset({MetadataField.SIZE, MetadataField.CONTENT_TYPE}) + document_families = frozenset({DocumentFamily.SCANNED_IMAGE, DocumentFamily.PDF}) + + def file_collection_inventory(self, collection): + raise AssertionError("capabilities must not aggregate anything") + + +class _DiscoveringSource(_AggregatingSource): + def list_file_collections(self): + raise AssertionError("capabilities must not list anything") + + +class _ResultSource(_AggregatingSource): + def quote_identifier(self, name): + raise AssertionError("capabilities must not build SQL") + + def source_sample_sql(self, collection, formats, sample_files): + raise AssertionError("capabilities must not build SQL") + + def run_file_aggregate(self, sql, measures): + raise AssertionError("capabilities must not execute anything") + + +class _Format: + name = ResultFormatName.MAPPED_COLUMNS + document_families = frozenset(DocumentFamily) + + def __init__(self, *connectors: str) -> None: + self.connectors = frozenset(connectors) + + def row_diagnostics(self, binding, quote): + raise AssertionError("capabilities must not build SQL") + + +def test_the_fakes_satisfy_exactly_the_tiers_they_are_meant_to(): + assert isinstance(_AggregatingSource(), FileCollectionSource) + assert not isinstance(_AggregatingSource(), DiscoveringFileSource) + assert not isinstance(_AggregatingSource(), FileResultSource) + assert isinstance(_DiscoveringSource(), DiscoveringFileSource) + assert isinstance(_ResultSource(), FileResultSource) + assert MeasureKind.COUNT.value == "count" + + +def test_an_aggregating_source_without_discovery_or_results_says_which_is_missing(): + capabilities = file_capabilities(_AggregatingSource()) + assert capabilities.metadata_aggregation.available is True + assert capabilities.collection_discovery.limitation is ( + CapabilityLimitation.NO_COLLECTION_DISCOVERY + ) + assert capabilities.result_assessment.limitation is ( + CapabilityLimitation.NO_RESULT_SOURCE + ) + # Declared order, whatever order the frozenset iterates in. + assert capabilities.document_families == ( + DocumentFamily.PDF, + DocumentFamily.SCANNED_IMAGE, + ) + + +def test_discovery_is_its_own_capability(): + capabilities = file_capabilities(_DiscoveringSource()) + assert capabilities.collection_discovery.available is True + assert capabilities.collection_discovery.limitation is None + + +def test_a_result_source_with_no_format_for_its_dialect_cannot_assess(): + capabilities = file_capabilities(_ResultSource(), (_Format("otherwh"),)) + assert capabilities.result_assessment.limitation is ( + CapabilityLimitation.NO_RESULT_FORMAT + ) + assert capabilities.result_formats == () + + +def test_a_result_source_with_a_format_for_its_dialect_can_assess(): + capabilities = file_capabilities( + _ResultSource(), (_Format("otherwh"), _Format("fakewh")) + ) + assert capabilities.result_assessment.available is True + assert capabilities.result_formats == (ResultFormatName.MAPPED_COLUMNS,) + + +def test_no_result_format_is_registered_yet(): + assert RESULT_FORMATS == () + assert file_capabilities(_ResultSource()).result_assessment.limitation is ( + CapabilityLimitation.NO_RESULT_FORMAT + ) + + +def test_a_capability_is_available_or_names_why_not(): + with pytest.raises(ValueError, match="no limitation"): + Capability(available=True, limitation=CapabilityLimitation.NO_FILE_SOURCE) + with pytest.raises(ValueError, match="name its limitation"): + Capability(available=False) + explained = Capability( + available=False, limitation=CapabilityLimitation.NO_FILE_SOURCE + ) + assert "object storage" in explained.explanation + + +# --- native processing ------------------------------------------------------------ + + +def test_native_processing_is_unavailable_with_a_specific_reason(): + native = NativeProcessing() + assert native.available is False + assert native.reason == "no_verified_hard_spend_cap" + assert "hard spend cap" in native.explanation + assert native.model_dump() == { + "available": False, + "reason": "no_verified_hard_spend_cap", + "explanation": native.explanation, + } + + +def test_nothing_can_report_native_processing_available(): + """No field exists to set, so construction and deserialization refuse the + key outright, and a copy that tries to write it still reports false.""" + + with pytest.raises(ValueError): + NativeProcessing(available=True) + with pytest.raises(ValueError): + NativeProcessing.model_validate({"available": True}) + copied = NativeProcessing().model_copy(update={"available": True}) + assert copied.available is False + assert copied.model_dump()["available"] is False + + payload = file_capabilities(_ResultSource(), (_Format("fakewh"),)).payload() + assert payload["native_processing"]["available"] is False + + +def test_the_capability_payload_is_plain_json_that_passes_the_sanitizer(): + payload = file_capabilities(_DiscoveringSource()).payload() + env.sanitize(env.ok({"files": payload})) + assert payload["metadata_aggregation"] == { + "available": True, + "limitation": None, + "explanation": None, + } + assert payload["document_families"] == ["pdf", "scanned_image"] + + +def test_limitation_codes_are_the_whole_vocabulary(): + """A new limitation needs engine-owned wording before it can be reported.""" + + from exmergo_dex_core.files.contract import LIMITATION_TEXT + + assert set(LIMITATION_TEXT) == set(Limitation) + assert all(text and "\n" not in text for text in LIMITATION_TEXT.values()) diff --git a/packages/dex-core/tests/files/test_file_results.py b/packages/dex-core/tests/files/test_file_results.py new file mode 100644 index 00000000..59b7d8a4 --- /dev/null +++ b/packages/dex-core/tests/files/test_file_results.py @@ -0,0 +1,623 @@ +"""The file aggregates: structurally incapable of carrying document content, and +never able to spell an absent measurement the way they spell a zero.""" + +from __future__ import annotations + +import enum +import json +import types +import typing +from datetime import UTC, datetime + +import pytest +from pydantic import ( + AwareDatetime, + Strict, + StringConstraints, + TypeAdapter, + ValidationError, +) + +from exmergo_dex_core import envelope as env +from exmergo_dex_core.files import results as file_results +from exmergo_dex_core.files.contract import ( + DIAGNOSTIC_BIN_EDGES, + CollectionKind, + Diagnostic, + DocumentFamily, + Limitation, + MetadataField, + ResultFormatName, + SamplingMethod, + UnavailableReason, +) +from exmergo_dex_core.files.results import ( + Bin, + CollectionInventory, + CollectionScope, + CollectionSummary, + Coverage, + Currency, + DiagnosticDistributions, + Distribution, + FileAggregate, + FileProfile, + FormatCounts, + Processing, + Ratio, + StatusCounts, + Unavailable, +) + +NOW = datetime(2026, 9, 10, 12, 0, tzinfo=UTC) +EARLIER = datetime(2026, 9, 1, 8, 30, tzinfo=UTC) + + +# --- the structural guarantee ------------------------------------------------- + + +def _aggregate_models(module=file_results) -> list[type[FileAggregate]]: + found, stack = [], [FileAggregate] + while stack: + for sub in stack.pop().__subclasses__(): + stack.append(sub) + if sub.__module__ == module.__name__: + found.append(sub) + return found + + +def _leaves(annotation, metadata): + """Every leaf type an annotation admits, with the constraints on it.""" + + origin = typing.get_origin(annotation) + if origin is typing.Annotated: + base, *extra = typing.get_args(annotation) + yield from _leaves(base, [*metadata, *extra]) + elif origin in (typing.Union, types.UnionType): + for arg in typing.get_args(annotation): + yield from _leaves(arg, []) + elif origin is tuple: + for arg in typing.get_args(annotation): + if arg is not Ellipsis: + yield from _leaves(arg, []) + else: + yield annotation, metadata + + +def _constraints(metadata): + """Constraints, including the ones pydantic nests inside a FieldInfo.""" + + for item in metadata: + yield item + yield from getattr(item, "metadata", ()) + + +def _admissible(leaf, metadata) -> bool: + constraints = list(_constraints(metadata)) + if leaf is int: + return any(isinstance(c, Strict) and c.strict for c in constraints) + if leaf is str: + return any( + isinstance(c, StringConstraints) + and c.pattern == file_results.RELATION_NAME_PATTERN + for c in constraints + ) + if leaf is AwareDatetime: + return True + return isinstance(leaf, type) and issubclass(leaf, (enum.Enum, FileAggregate)) + + +def _inadmissible_fields(model: type[FileAggregate]) -> list[str]: + return sorted( + name + for name, info in model.model_fields.items() + if not all( + _admissible(leaf, meta) + for leaf, meta in _leaves(info.annotation, info.metadata) + ) + ) + + +def test_the_walker_sees_every_public_aggregate(): + names = {m.__name__ for m in _aggregate_models()} + assert names >= { + "Unavailable", + "Ratio", + "FormatCounts", + "CollectionSummary", + "CollectionInventory", + "Bin", + "Distribution", + "DiagnosticDistributions", + "StatusCounts", + "CollectionScope", + "Coverage", + "Currency", + "Processing", + "FileProfile", + } + + +@pytest.mark.parametrize("model", _aggregate_models(), ids=lambda m: m.__name__) +def test_no_aggregate_field_can_hold_free_text_none_or_a_loose_number(model): + """Every leaf is a strict count, a fixed category, a timestamp, another + aggregate, or a relation name, so there is no field an excerpt, a file path, + or a provider's error body could be written into, and none that reads an + absent measurement as ``None``. The next person to add ``uri: str`` meets + this test.""" + + assert _inadmissible_fields(model) == [] + + +def test_the_walker_catches_the_fields_it_exists_to_catch(): + class Leaky(FileAggregate): + uri: str + note: str | None + pages: int + extra: dict + fine: file_results.NonNegative + + Leaky.__module__ = "tests.leaky" + assert _inadmissible_fields(Leaky) == ["extra", "note", "pages", "uri"] + + +@pytest.mark.parametrize( + "hostile", + [ + "gs://exmergo-docs/2026/CANARY-FILENAME-jane-doe.pdf", + "https://storage.googleapis.com/b/o.pdf?X-Goog-Signature=abc&X-Goog-Expires=600", + "/Volumes/main/raw/docs/contract.pdf", + "@stage/docs/contract.pdf", + "CANARY-DOC-TEXT Total due 54.00", + "INVALID_ARGUMENT: could not read 'Jane Doe'", + "proj.dataset.table; DROP TABLE x", + "`proj.dataset.table`", + "", + ], +) +def test_a_relation_name_refuses_paths_urls_text_and_quoting(hostile): + with pytest.raises(ValidationError): + TypeAdapter(file_results.RelationName).validate_python(hostile) + + +@pytest.mark.parametrize( + "name", ["exmergo-viz.dex_ci.files_obj", "RAW.DOCS.INVOICES", "main.docs"] +) +def test_a_relation_name_accepts_the_identifiers_connectors_report(name): + assert TypeAdapter(file_results.RelationName).validate_python(name) == name + + +def test_an_aggregate_refuses_a_key_it_does_not_declare(): + payload = _inventory().model_dump(mode="json") + payload["uri"] = "gs://bucket/secret.pdf" + with pytest.raises(ValidationError, match="uri"): + CollectionInventory.model_validate(payload) + + +def test_an_aggregate_cannot_be_edited_after_it_is_built(): + inventory = _inventory() + with pytest.raises(ValidationError): + inventory.file_count = 3 + + +# --- absent versus zero ------------------------------------------------------- + + +@pytest.mark.parametrize("value", [None, "12", True, -1, 1.0]) +def test_a_count_refuses_anything_but_a_non_negative_integer(value): + with pytest.raises(ValidationError): + TypeAdapter(file_results.Count).validate_python(value) + + +def test_an_observed_zero_and_an_unavailable_count_serialize_differently(): + counts = TypeAdapter(file_results.Count) + assert counts.dump_python(0, mode="json") == 0 + unavailable = Unavailable(reason=UnavailableReason.NOT_REPORTED) + assert counts.dump_python(unavailable, mode="json") == {"reason": "not_reported"} + + +def test_an_instant_is_a_timezone_aware_moment_or_unavailable(): + instants = TypeAdapter(file_results.Instant) + assert instants.validate_python(NOW) == NOW + with pytest.raises(ValidationError): + instants.validate_python(datetime(2026, 9, 10)) + with pytest.raises(ValidationError): + instants.validate_python(None) + + +def test_a_ratio_carries_both_sides_and_stays_within_its_denominator(): + assert Ratio(numerator=3, denominator=4).payload() == { + "numerator": 3, + "denominator": 4, + "fraction": 0.75, + } + assert Ratio(numerator=0, denominator=0).fraction is None + with pytest.raises(ValidationError, match="cannot exceed"): + Ratio(numerator=5, denominator=4) + + +# --- inventory ------------------------------------------------------------------ + + +def _inventory(**overrides) -> CollectionInventory: + fields = { + "collection": "exmergo-viz.dex_ci.files_obj", + "kind": CollectionKind.BIGQUERY_OBJECT_TABLE, + "observed_at": NOW, + "file_count": 21, + "known_bytes": 91_234, + "missing_size": 0, + "zero_byte": 1, + "formats": FormatCounts(pdf=12, jpeg=2, png=3, tiff=2, other=1, unknown=1), + "earliest_update": EARLIER, + "latest_update": NOW, + "metadata_refreshed_at": Unavailable(reason=UnavailableReason.NOT_REPORTED), + "limitations": ( + Limitation.FORMAT_IS_REPORTED_METADATA, + Limitation.COLLECTION_IS_REGISTERED_INDEX, + ), + } + fields.update(overrides) + return CollectionInventory(**fields) + + +def test_an_inventory_round_trips_through_json(): + inventory = _inventory() + assert ( + CollectionInventory.model_validate_json(inventory.model_dump_json()) + == inventory + ) + assert CollectionInventory.model_validate( + json.loads(inventory.model_dump_json()) + ) == (inventory) + + +def test_an_inventory_counts_every_file_in_exactly_one_format_bucket(): + with pytest.raises(ValidationError, match="format buckets sum to 20"): + _inventory( + formats=FormatCounts(pdf=11, jpeg=2, png=3, tiff=2, other=1, unknown=1) + ) + + +def test_an_inventory_cannot_have_more_zero_byte_files_than_sized_ones(): + with pytest.raises(ValidationError, match="zero bytes"): + _inventory(missing_size=20, zero_byte=2) + + +def test_an_inventory_with_no_size_metadata_says_so_instead_of_reporting_zero(): + unreported = Unavailable(reason=UnavailableReason.NOT_REPORTED) + inventory = _inventory( + known_bytes=unreported, missing_size=unreported, zero_byte=unreported + ) + payload = inventory.payload() + assert payload["known_bytes"] == {"reason": "not_reported"} + assert payload["zero_byte"] == {"reason": "not_reported"} + + +def test_an_inventory_states_its_limitations(): + with pytest.raises(ValidationError, match="collection_is_registered_index"): + _inventory(limitations=(Limitation.FORMAT_IS_REPORTED_METADATA,)) + with pytest.raises(ValidationError, match="twice"): + _inventory( + limitations=( + Limitation.FORMAT_IS_REPORTED_METADATA, + Limitation.COLLECTION_IS_REGISTERED_INDEX, + Limitation.FORMAT_IS_REPORTED_METADATA, + ) + ) + + +def test_an_inventory_update_range_runs_forwards(): + with pytest.raises(ValidationError, match="earliest update"): + _inventory(earliest_update=NOW, latest_update=EARLIER) + + +def test_discovery_reports_a_count_it_did_not_scan_for_as_unavailable(): + summary = CollectionSummary( + collection="exmergo-viz.dex_ci.files_obj", + kind=CollectionKind.BIGQUERY_OBJECT_TABLE, + file_count=Unavailable(reason=UnavailableReason.NOT_REPORTED), + metadata_refreshed_at=Unavailable(reason=UnavailableReason.NOT_REPORTED), + ) + assert summary.payload()["file_count"] == {"reason": "not_reported"} + + +# --- profile -------------------------------------------------------------------- + + +def _distribution( + diagnostic: Diagnostic, values: list[int], *, absent: int = 0, invalid: int = 0 +) -> Distribution: + edges = DIAGNOSTIC_BIN_EDGES[diagnostic] + files = [0] * len(edges) + for value in values: + files[max(i for i, edge in enumerate(edges) if value >= edge)] += 1 + no_evidence = Unavailable(reason=UnavailableReason.NO_EVIDENCE) + return Distribution( + assessed=len(values) + absent + invalid, + files_with_value=len(values), + files_absent=absent, + files_invalid=invalid, + minimum=min(values) if values else no_evidence, + maximum=max(values) if values else no_evidence, + bins=tuple(Bin(at_least=e, files=n) for e, n in zip(edges, files, strict=True)), + ) + + +def _diagnostics(**overrides) -> DiagnosticDistributions: + unsupported = Unavailable(reason=UnavailableReason.NOT_SUPPORTED_BY_FORMAT) + fields = { + # Seven matched files: five with text (one an observed empty string), a + # failure with no payload, and one malformed payload. + "text_characters": _distribution( + Diagnostic.TEXT_CHARACTERS, [2400, 400, 0, 150, 600], absent=1, invalid=1 + ), + "reported_pages": _distribution( + Diagnostic.REPORTED_PAGES, [3, 1, 1, 1, 2], absent=1, invalid=1 + ), + "represented_pages": _distribution( + Diagnostic.REPRESENTED_PAGES, [3, 1, 1, 1, 1], absent=1, invalid=1 + ), + "tables": _distribution( + Diagnostic.TABLES, [1, 0, 0, 0, 0], absent=1, invalid=1 + ), + "form_fields": unsupported, + "paragraphs": unsupported, + } + fields.update(overrides) + return DiagnosticDistributions(**fields) + + +def _profile(**overrides) -> FileProfile: + fields = { + "collection": CollectionScope( + collection="exmergo-viz.dex_ci.files_obj", + kind=CollectionKind.BIGQUERY_OBJECT_TABLE, + result_table="exmergo-viz.dex_ci.files_docai_results", + result_format=ResultFormatName.BIGQUERY_DOCUMENT_AI, + families=(DocumentFamily.PDF, DocumentFamily.SCANNED_IMAGE), + metadata_fields=(MetadataField.SIZE, MetadataField.VERSION), + ), + "observed_at": NOW, + "coverage": Coverage( + requested=200, + eligible=10, + sampled=10, + sampling_method=SamplingMethod.SOURCE_IDENTITY_HASH, + matched=Ratio(numerator=7, denominator=10), + missing_result=Ratio(numerator=2, denominator=10), + ambiguous=Ratio(numerator=1, denominator=10), + ), + "currency": Currency( + version_matched=Ratio(numerator=5, denominator=7), + version_mismatched=Ratio(numerator=1, denominator=7), + version_unknown=Ratio(numerator=1, denominator=7), + ), + "processing": Processing( + statuses=StatusCounts(success=5, failure=1, partial=1, unknown=0), + payload_valid=Ratio(numerator=5, denominator=7), + partial_page_files=1, + diagnostics=_diagnostics(), + ), + "limitations": ( + Limitation.DOCUMENT_PII_NOT_SCREENED, + Limitation.NATIVE_PROCESSING_UNAVAILABLE, + Limitation.SAMPLING_NOT_REPRESENTATIVE, + Limitation.PAGES_PARTIALLY_REPRESENTED, + ), + } + fields.update(overrides) + return FileProfile(**fields) + + +def test_a_profile_round_trips_through_json(): + profile = _profile() + assert FileProfile.model_validate_json(profile.model_dump_json()) == profile + assert FileProfile.model_validate(json.loads(profile.model_dump_json())) == profile + + +def test_a_profile_and_an_inventory_pass_the_envelope_sanitizer(): + envelope = env.ok( + {"profile": _profile().payload(), "inventory": _inventory().payload()} + ) + assert env.sanitize(envelope) is envelope + + +def test_the_diagnostics_model_and_the_diagnostic_vocabulary_stay_in_step(): + assert set(DiagnosticDistributions.model_fields) == {d.value for d in Diagnostic} + + +def test_files_without_a_result_can_never_leave_the_denominator(): + with pytest.raises(ValidationError, match="stays in the denominator"): + Coverage( + requested=200, + eligible=10, + sampled=10, + sampling_method=SamplingMethod.SOURCE_IDENTITY_HASH, + matched=Ratio(numerator=7, denominator=10), + missing_result=Ratio(numerator=0, denominator=10), + ambiguous=Ratio(numerator=1, denominator=10), + ) + + +def test_coverage_rates_share_the_sample_as_their_denominator(): + with pytest.raises(ValidationError, match="over the sampled files"): + Coverage( + requested=200, + eligible=10, + sampled=10, + sampling_method=SamplingMethod.SOURCE_IDENTITY_HASH, + matched=Ratio(numerator=7, denominator=7), + missing_result=Ratio(numerator=2, denominator=10), + ambiguous=Ratio(numerator=1, denominator=10), + ) + + +@pytest.mark.parametrize( + ("requested", "eligible", "sampled"), [(5, 10, 10), (200, 10, 5)] +) +def test_a_sample_takes_every_eligible_file_up_to_the_requested_size( + requested, eligible, sampled +): + with pytest.raises(ValidationError, match="smaller of the two"): + Coverage( + requested=requested, + eligible=eligible, + sampled=sampled, + sampling_method=SamplingMethod.SOURCE_IDENTITY_HASH, + matched=Ratio(numerator=sampled, denominator=sampled), + missing_result=Ratio(numerator=0, denominator=sampled), + ambiguous=Ratio(numerator=0, denominator=sampled), + ) + + +@pytest.mark.parametrize("requested", [0, 1001]) +def test_a_sample_request_stays_within_the_ceiling(requested): + with pytest.raises(ValidationError): + Coverage( + requested=requested, + eligible=0, + sampled=0, + sampling_method=SamplingMethod.SOURCE_IDENTITY_HASH, + matched=Ratio(numerator=0, denominator=0), + missing_result=Ratio(numerator=0, denominator=0), + ambiguous=Ratio(numerator=0, denominator=0), + ) + + +def test_currency_splits_the_matched_files_three_ways(): + with pytest.raises(ValidationError, match="must add up"): + Currency( + version_matched=Ratio(numerator=5, denominator=7), + version_mismatched=Ratio(numerator=1, denominator=7), + version_unknown=Ratio(numerator=0, denominator=7), + ) + + +def test_a_profile_measures_currency_over_the_matched_files(): + with pytest.raises(ValidationError, match="currency"): + _profile( + currency=Currency( + version_matched=Ratio(numerator=8, denominator=10), + version_mismatched=Ratio(numerator=1, denominator=10), + version_unknown=Ratio(numerator=1, denominator=10), + ) + ) + + +def test_statuses_count_the_matched_files_only(): + processing = _profile().processing + with pytest.raises(ValidationError, match="statuses"): + _profile( + processing=processing.model_copy( + update={ + "statuses": StatusCounts(success=6, failure=1, partial=1, unknown=0) + } + ) + ) + + +def test_ambiguous_duplicates_stay_out_of_every_content_distribution(): + """Eight assessed would mean the ambiguous file was arbitrarily resolved.""" + + processing = _profile().processing + widened = _diagnostics( + text_characters=_distribution( + Diagnostic.TEXT_CHARACTERS, + [2400, 400, 0, 150, 600, 90], + absent=1, + invalid=1, + ) + ) + with pytest.raises(ValidationError, match="excludes ambiguous"): + _profile(processing=processing.model_copy(update={"diagnostics": widened})) + + +def test_a_profile_states_what_it_did_not_establish(): + with pytest.raises(ValidationError, match="document_pii_not_screened"): + _profile( + limitations=( + Limitation.NATIVE_PROCESSING_UNAVAILABLE, + Limitation.SAMPLING_NOT_REPRESENTATIVE, + ) + ) + + +def test_a_profile_has_no_readiness_score(): + fields = set(FileProfile.model_fields) + assert fields == { + "collection", + "observed_at", + "coverage", + "currency", + "processing", + "limitations", + } + assert not any("score" in f or "ready" in f for f in fields) + + +# --- distributions ---------------------------------------------------------------- + + +def test_a_distribution_partitions_its_assessed_files(): + with pytest.raises(ValidationError, match="sum to 6"): + Distribution( + assessed=7, + files_with_value=5, + files_absent=1, + files_invalid=0, + minimum=0, + maximum=3, + bins=(Bin(at_least=0, files=1), Bin(at_least=1, files=4)), + ) + + +def test_an_observed_zero_lands_in_the_first_bin_and_an_absence_does_not(): + distribution = _distribution(Diagnostic.TABLES, [0, 0, 1], absent=2) + assert distribution.bins[0].files == 2 + assert distribution.files_absent == 2 + assert distribution.minimum == 0 + + +def test_with_no_values_there_is_no_minimum_or_maximum(): + empty = _distribution(Diagnostic.TABLES, [], absent=3) + assert empty.minimum == Unavailable(reason=UnavailableReason.NO_EVIDENCE) + with pytest.raises(ValidationError, match="no minimum or maximum"): + Distribution( + assessed=3, + files_with_value=0, + files_absent=3, + files_invalid=0, + minimum=0, + maximum=0, + bins=(Bin(at_least=0, files=0),), + ) + + +def test_bins_start_at_zero_rise_and_hold_exactly_the_valued_files(): + with pytest.raises(ValidationError, match="start at 0"): + Distribution( + assessed=1, + files_with_value=1, + files_absent=0, + files_invalid=0, + minimum=2, + maximum=2, + bins=(Bin(at_least=1, files=1),), + ) + with pytest.raises(ValidationError, match="exactly the files"): + Distribution( + assessed=2, + files_with_value=2, + files_absent=0, + files_invalid=0, + minimum=1, + maximum=2, + bins=(Bin(at_least=0, files=0), Bin(at_least=1, files=1)), + ) + + +def test_each_diagnostic_uses_its_own_fixed_bins(): + wrong = _distribution(Diagnostic.TEXT_CHARACTERS, [3, 1], absent=0) + with pytest.raises(ValidationError, match="fixed bin edges"): + _diagnostics(reported_pages=wrong) From 0fa96f6b8112fb4f710ac20201d3cad98b1632f3 Mon Sep 17 00:00:00 2001 From: Marco Ciavarella Date: Thu, 10 Sep 2026 14:35:56 +0200 Subject: [PATCH 2/4] Add safety_spine tests for file contracts and results (BigQuery) --- packages/dex-core/tests/test_safety_spine.py | 67 ++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/packages/dex-core/tests/test_safety_spine.py b/packages/dex-core/tests/test_safety_spine.py index 3d20ffe2..f7102f40 100644 --- a/packages/dex-core/tests/test_safety_spine.py +++ b/packages/dex-core/tests/test_safety_spine.py @@ -3405,6 +3405,73 @@ def test_no_payload_is_keyed_by_a_warehouse_object_name(capsys): assert capsys.readouterr().out +_FILE_CANARIES = [ + "gs://exmergo-docs/2026/CANARY-FILENAME-jane-doe-078-05-1120.pdf", + "https://storage.googleapis.com/b/o.pdf?X-Goog-Signature=abc&X-Goog-Expires=600", + "CANARY-DOC-TEXT Total due 54.00", + "INVALID_ARGUMENT: CANARY-PROVIDER-ERROR could not read 'Jane Doe'", +] + + +def test_file_aggregates_cannot_carry_document_content(): + """File exploration returns aggregates only, and the types enforce it rather + than trusting every caller to comply: the one string type refuses anything + shaped like a path, a URL, extracted text, or a provider's error body, and no + aggregate accepts a key it does not declare. ``tests/files/ + test_file_results.py`` walks every field of every aggregate; this is the + spine's statement of the rule.""" + + from pydantic import TypeAdapter, ValidationError + + from exmergo_dex_core.files.results import CollectionSummary, RelationName + + names = TypeAdapter(RelationName) + summary = { + "collection": "exmergo-viz.dex_ci.files_obj", + "kind": "bigquery_object_table", + "file_count": 0, + "metadata_refreshed_at": {"reason": "not_reported"}, + } + CollectionSummary.model_validate(summary) + for canary in _FILE_CANARIES: + with pytest.raises(ValidationError): + names.validate_python(canary) + with pytest.raises(ValidationError): + CollectionSummary.model_validate({**summary, "collection": canary}) + for key in ("uri", "path", "text", "error", "status_detail"): + with pytest.raises(ValidationError): + CollectionSummary.model_validate({**summary, key: canary}) + + +@pytest.mark.parametrize( + "connector", + [ + "duckdb", + "bigquery", + "snowflake", + "databricks", + "postgres", + "redshift", + "clickhouse", + ], +) +def test_no_connector_can_report_native_document_processing(connector): + """New document-processing charges need a provider-enforced hard spend cap + dex can verify, and none exists, so native processing is unavailable on every + connector and cannot be made available by construction or deserialization.""" + + from exmergo_dex_core.adapters import _adapter_class + from exmergo_dex_core.files.contract import NativeProcessing, file_capabilities + + capabilities = file_capabilities(object.__new__(_adapter_class(connector))) + assert capabilities.native_processing.available is False + assert capabilities.payload()["native_processing"]["available"] is False + with pytest.raises(ValueError): + NativeProcessing(available=True) + with pytest.raises(ValueError): + NativeProcessing.model_validate({"available": True}) + + # --- BigQuery: the billed connector exercises every family --------------------- # # These run against the fake client (tests/fakes/bigquery.py): deterministic, From 13015e1105f4f80abac7f635bf2d5c4be5b1b061 Mon Sep 17 00:00:00 2001 From: Marco Ciavarella Date: Thu, 10 Sep 2026 14:39:08 +0200 Subject: [PATCH 3/4] Update docs --- CHANGELOG.md | 32 ++++++ references/file-exploration.md | 198 +++++++++++++++++++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 references/file-exploration.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 4558896b..b2f25131 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,38 @@ tag releases both in lockstep, so entries below are keyed by the engine version. ## [Unreleased] +### Added + +- **The file-exploration contract, with no command yet** ([#450]). A new + `exmergo_dex_core.files` package defines how a document collection (a BigQuery + object table, a Snowflake directory table, a Databricks volume or manifest) and + the processing results already materialized about it will be read. It holds: + - the fixed vocabularies: document families, format buckets, processing + statuses, and limitation codes + - optional connector protocols + - the request types + - a capability model + - the aggregate result types the coming `explore files` commands return + + The protocols sit beside the warehouse adapter rather than inside it, and they + are tiered: metadata aggregation, then result assessment, with collection + discovery as its own optional capability. A connector that implements none of + them reports each capability unavailable with a named reason, and no shipped + connector implements one yet. + + The result types are built so that they cannot carry document content: + - every field is a strict count, a fixed category, a timestamp, or a relation + name that refuses anything shaped like a path or a URL + - unknown keys are refused + - no field admits `None`, so an absent measurement is always an explicit + `{"reason": ...}` and never confused with an observed zero + + Native document processing is reported unavailable with the specific reason: + no supported path offers a provider-enforced hard spend cap that dex can + verify. It has no settable field, so no configuration or confirmation can + report it available. Nothing is exported from the package root yet, and no + existing behavior changes. + ## [1.12.2] - 2026-09-09 ### Fixed diff --git a/references/file-exploration.md b/references/file-exploration.md new file mode 100644 index 00000000..155c4503 --- /dev/null +++ b/references/file-exploration.md @@ -0,0 +1,198 @@ +# File exploration: the contract + +**Status.** The contract exists in the engine as the `exmergo_dex_core.files` +package. No command uses it yet, and no connector implements it yet, so every +connector reports file exploration unavailable with a named reason. This page +describes the contract that the `explore files` commands and the connector +implementations will be built on. It is the reference for anyone implementing or +reviewing them. + +## What it is for + +A file collection is an index of documents in object storage: a BigQuery object +table, a Snowflake directory table, a Databricks volume, or a table listing files. +Teams that turn such a collection into staging and intermediate models usually +already have a table of processing results, written by an earlier pipeline that +ran a document parser. File exploration answers questions about those two things +together: + +- What files the collection holds. +- How many of them have a processing result. +- Whether those results describe the files as they are now. +- Where processing failed or produced little usable structure. +- What the evidence does not establish. + +## What it never does + +- **It never reads document content out of the warehouse.** Every content + computation is a warehouse expression, and only counts come back. Nothing that + leaves the warehouse carries any of the following: + - a document body, excerpt, heading, or extracted value + - an image + - a file name, path, signed URL, or individual file identifier + - a provider's error text +- **It never invokes document processing.** Dex reads results that already exist. + New processing charges would need a provider-enforced hard spend cap that + covers the exact operation, holds against concurrent and in-flight work, and + that dex can verify. No supported processing path offers one today. Estimates, + page limits, delayed quotas, and user confirmation do not substitute for one, + so native processing is reported unavailable everywhere and has no setting + that changes that. +- **It never creates or refreshes anything.** Collections and result tables are + read as configured. Dex does not create an object table, a stage, a volume, or + a connection, and it does not refresh external metadata to improve its answer. +- **It never falls back.** A connector without a file source says so. Dex does + not switch to another connector, list the bucket itself, or download a file. + +## Formats are reported metadata + +Files are counted into fixed buckets: `pdf`, `jpeg`, `png`, `tiff`, `other`, +`unknown`. The bucket comes from the content type the storage layer recorded, +never from the file's bytes and never from its name. A PNG uploaded under a `.pdf` +name counts as a PNG. + +- A well-formed content type outside the supported families is `other`. +- A missing or malformed content type is `unknown`, and so is one that states the + format is unknown (`application/octet-stream`). +- Parameters and case are ignored: `Application/PDF; version=1.7` is `pdf`. + +Two document families can be assessed: `pdf`, and `scanned_image`, which covers +`jpeg`, `png`, and `tiff`. The second name describes a document-image input. It +does not claim dex checked that an image depicts a scanned document. + +## Absent is not zero + +Every aggregate field is either an observed number or an explicit unavailability +with its reason. No field is ever `null`: + +```json +{"zero_byte": 0} +{"zero_byte": {"reason": "not_reported"}} +``` + +The first says no file is empty. The second says the collection's metadata does +not carry sizes, so emptiness was not measured. The reasons are: + +- `not_reported` +- `not_bound` +- `not_supported_by_format` +- `unrecognized_shape` +- `no_evidence` +- `not_assessed` + +Every rate carries both sides, for example +`{"numerator": 5, "denominator": 7, "fraction": 0.714}`. + +## Capabilities + +A connector's file capabilities are read off the protocols it implements. It +never declares them with a flag, so it cannot claim one it does not have. The +report has three capabilities and two lists: + +| Entry | Meaning | +|---|---| +| `collection_discovery` | lists the collections inside the source scope from catalog metadata, without scanning any of them | +| `metadata_aggregation` | counts one collection's files, bytes, formats, and update times in one budgeted statement | +| `result_assessment` | matches a sample of the collection against a materialized results table | +| `document_families` | the families the connector's metadata can bucket | +| `result_formats` | the result formats written for the connector's dialect | + +Each capability is either `{"available": true}` or unavailable with a named +limitation and a fixed explanation: + +- `no_file_source` +- `no_collection_discovery` +- `no_result_source` +- `no_result_format` + +`native_processing` is always `{"available": false, "reason": +"no_verified_hard_spend_cap"}`. + +## Result bindings + +A result table is never guessed from a similar name. It is bound to its +collection explicitly, and a binding names identifiers only. It never carries an +expression, a template, or a callback, so it cannot be used to run SQL. A binding +names: + +- the collection and the materialized result table +- the column holding each result's source-file identity +- the result format +- where the format finds its evidence: + - the stored parser payload (and status column) for a provider's native output + - or plain diagnostic columns for a pipeline that kept only those +- optionally, a column with the processed file's version, and a processing + timestamp + +The formats dex knows are: + +- `bigquery_document_ai` +- `snowflake_ai_parse_document` +- `databricks_ai_parse_document` +- `mapped_columns` + +A result source must be a materialized table. A view or table function can hide +a call that spends money and returns content, so neither is accepted. + +## What a profile reports + +A profile assesses a deterministic sample of the collection: 200 files by +default, and at most 1,000. Files are ordered by a hash of their source identity +with the identity as the tie-breaker, and chosen from the collection before any +result is matched, so a file with no result stays in the sample. The sample +bounds the assessment, not the warehouse scan, which is priced before it runs. It +is reproducible for unchanged input and is not a claim of statistical +representativeness. + +The report has five groups: + +| Group | Contents | +|---|---| +| `collection` | the collection, the result table and format, the families assessed, the metadata available | +| `coverage` | the sample split three ways: one resolvable result (`matched`), no result, or duplicates that could not be resolved to one (`ambiguous`) | +| `currency` | of the matched files, how many results match the file's current version, how many describe another version, and how many cannot be compared | +| `processing` | statuses (`success`, `failure`, `partial`, `unknown`), payload validity, files with only some pages represented, and a distribution per diagnostic | +| `limitations` | fixed statements of what the profile does not establish | + +The profile obeys these rules: + +- **Coverage always adds back up to the sample.** +- **Currency is compared only where both sides carry compatible version + evidence.** A processing timestamp alone does not prove which version was + processed, so everything else is reported unknown, never current. +- **Ambiguous duplicates are never picked.** They are excluded from every content + distribution. +- **Diagnostics have fixed bins.** The diagnostics are text characters, reported + pages, pages represented in the result, tables, form fields, and paragraphs. The + first bin holds exactly the observed zeros, and files with no value are counted + separately. +- **Stored status values are never returned.** Any value outside the fixed + status vocabulary is counted as `unknown`. + +Every profile states that document content was not screened for personal data, +that no document processing was invoked, and that the sample is not +representative. There is no overall readiness score. Successful parsing, long +text, or detected tables do not make an extraction correct. + +## Implementing a file source + +A connector reaches a capability by implementing the matching protocol in +`exmergo_dex_core.files.contract`: + +- **`FileCollectionSource`** for metadata aggregation. + - It declares `name`, `collection_kind`, `metadata_fields`, and + `document_families`. + - It implements `file_collection_inventory(collection)`. +- **`DiscoveringFileSource`** for discovery, with `list_file_collections()`. +- **`FileResultSource`** for result assessment. It adds: + - `quote_identifier` + - `source_sample_sql`, the bounded, deterministic source selection + - `run_file_aggregate`, which runs one engine-assembled aggregate statement + through the adapter's own cost gate and returns only the declared aliases + +A result format implements `ResultFormat`, which turns a binding into +per-row SQL expressions and never executes anything. + +Only the aggregate types in `exmergo_dex_core.files.results` cross from a source +into the command layer. Those types refuse free text, unknown keys, and `None` by +construction, so a source cannot return content even by mistake. From 6854bbd8a1a6917db6153e8bfc68bfda63b68c6f Mon Sep 17 00:00:00 2001 From: Marco Ciavarella Date: Thu, 10 Sep 2026 15:02:26 +0200 Subject: [PATCH 4/4] hotfix naming --- packages/dex-core/tests/files/test_file_results.py | 12 ++++++------ packages/dex-core/tests/test_safety_spine.py | 4 ++-- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/packages/dex-core/tests/files/test_file_results.py b/packages/dex-core/tests/files/test_file_results.py index 59b7d8a4..880860ea 100644 --- a/packages/dex-core/tests/files/test_file_results.py +++ b/packages/dex-core/tests/files/test_file_results.py @@ -164,7 +164,7 @@ class Leaky(FileAggregate): @pytest.mark.parametrize( "hostile", [ - "gs://exmergo-docs/2026/CANARY-FILENAME-jane-doe.pdf", + "gs://example-bucket/2026/CANARY-FILENAME-personal.pdf", "https://storage.googleapis.com/b/o.pdf?X-Goog-Signature=abc&X-Goog-Expires=600", "/Volumes/main/raw/docs/contract.pdf", "@stage/docs/contract.pdf", @@ -181,7 +181,7 @@ def test_a_relation_name_refuses_paths_urls_text_and_quoting(hostile): @pytest.mark.parametrize( - "name", ["exmergo-viz.dex_ci.files_obj", "RAW.DOCS.INVOICES", "main.docs"] + "name", ["my-project.docs.files_obj", "RAW.DOCS.INVOICES", "main.docs"] ) def test_a_relation_name_accepts_the_identifiers_connectors_report(name): assert TypeAdapter(file_results.RelationName).validate_python(name) == name @@ -241,7 +241,7 @@ def test_a_ratio_carries_both_sides_and_stays_within_its_denominator(): def _inventory(**overrides) -> CollectionInventory: fields = { - "collection": "exmergo-viz.dex_ci.files_obj", + "collection": "my-project.docs.files_obj", "kind": CollectionKind.BIGQUERY_OBJECT_TABLE, "observed_at": NOW, "file_count": 21, @@ -314,7 +314,7 @@ def test_an_inventory_update_range_runs_forwards(): def test_discovery_reports_a_count_it_did_not_scan_for_as_unavailable(): summary = CollectionSummary( - collection="exmergo-viz.dex_ci.files_obj", + collection="my-project.docs.files_obj", kind=CollectionKind.BIGQUERY_OBJECT_TABLE, file_count=Unavailable(reason=UnavailableReason.NOT_REPORTED), metadata_refreshed_at=Unavailable(reason=UnavailableReason.NOT_REPORTED), @@ -371,9 +371,9 @@ def _diagnostics(**overrides) -> DiagnosticDistributions: def _profile(**overrides) -> FileProfile: fields = { "collection": CollectionScope( - collection="exmergo-viz.dex_ci.files_obj", + collection="my-project.docs.files_obj", kind=CollectionKind.BIGQUERY_OBJECT_TABLE, - result_table="exmergo-viz.dex_ci.files_docai_results", + result_table="my-project.docs.parsed_documents", result_format=ResultFormatName.BIGQUERY_DOCUMENT_AI, families=(DocumentFamily.PDF, DocumentFamily.SCANNED_IMAGE), metadata_fields=(MetadataField.SIZE, MetadataField.VERSION), diff --git a/packages/dex-core/tests/test_safety_spine.py b/packages/dex-core/tests/test_safety_spine.py index f7102f40..782d35a6 100644 --- a/packages/dex-core/tests/test_safety_spine.py +++ b/packages/dex-core/tests/test_safety_spine.py @@ -3406,7 +3406,7 @@ def test_no_payload_is_keyed_by_a_warehouse_object_name(capsys): _FILE_CANARIES = [ - "gs://exmergo-docs/2026/CANARY-FILENAME-jane-doe-078-05-1120.pdf", + "gs://example-bucket/2026/CANARY-FILENAME-personal.pdf", "https://storage.googleapis.com/b/o.pdf?X-Goog-Signature=abc&X-Goog-Expires=600", "CANARY-DOC-TEXT Total due 54.00", "INVALID_ARGUMENT: CANARY-PROVIDER-ERROR could not read 'Jane Doe'", @@ -3427,7 +3427,7 @@ def test_file_aggregates_cannot_carry_document_content(): names = TypeAdapter(RelationName) summary = { - "collection": "exmergo-viz.dex_ci.files_obj", + "collection": "my-project.docs.files_obj", "kind": "bigquery_object_table", "file_count": 0, "metadata_refreshed_at": {"reason": "not_reported"},