# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-embedding-metadata-invariants-must-survive-schema-evolution-in-vector-indices.md # Remnant Warrant: https://corpusfetch.com/c/what-embedding-metadata-invariants-must-survive-schema-evolution-in-vector-indices.warrant.json # Markdown SHA-256: 38abab2c9abd0facaaf0afd98f68c32b8bb6effed93471ae16dbe9b34dae75c5 # Code AST SHA-256: 413e373313dbda362cd9003104c80a73edff01dc84677b335d478a42ca265669 # Verified Classes: MetadataVerdict, VectorMetadata, SchemaContext, MetadataPolicy, MetadataManifest, EmbeddingMetadataGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class MetadataVerdict(Enum): SINGLE_VERSION = auto() CROSS_VERSION_TRANSLATED = auto() NO_TRANSLATION_MATRIX = auto() UNVERIFIED_VERSION = auto() @dataclass(frozen=True) class VectorMetadata: vector_id: str schema_version: str schema_signature: str | None @dataclass(frozen=True) class SchemaContext: vectors: tuple[VectorMetadata, ...] current_query_schema_version: str field_mapping_available: bool @dataclass(frozen=True) class MetadataPolicy: require_translation_for_cross_version: bool @dataclass(frozen=True) class MetadataManifest: verdict: MetadataVerdict directive: str manifest_digest: str class EmbeddingMetadataGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_metadata( cls, ctx: SchemaContext, policy: MetadataPolicy ) -> MetadataManifest: # Gate 1: Signature presence for v in ctx.vectors: if not v.schema_signature: return MetadataManifest( verdict=MetadataVerdict.UNVERIFIED_VERSION, directive=f"HALT: Vector {v.vector_id} lacks schema version signature.", manifest_digest=cls._hash(f"NO_SIG:{v.vector_id}") ) # Gate 2: Version diversity check versions = {v.schema_version for v in ctx.vectors} # Gate 3: Single version if len(versions) == 1 and ctx.current_query_schema_version in versions: digest = cls._hash(json.dumps({ "version": ctx.current_query_schema_version, "vectors": str(len(ctx.vectors)) }, sort_keys=True)) return MetadataManifest( verdict=MetadataVerdict.SINGLE_VERSION, directive=f"PROCEED: All vectors are schema version {ctx.current_query_schema_version}.", manifest_digest=digest ) # Gate 4: Cross-version translation if policy.require_translation_for_cross_version and not ctx.field_mapping_available: return MetadataManifest( verdict=MetadataVerdict.NO_TRANSLATION_MATRIX, directive=f"HALT: Vectors span {len(versions)} schema versions, no translation matrix.", manifest_digest=cls._hash(f"NO_MATRIX:{len(versions)}") ) # Gate 5: Translated digest = cls._hash(json.dumps({ "versions": tuple(sorted(versions)), "query_version": ctx.current_query_schema_version, "vectors": str(len(ctx.vectors)), "translation": str(ctx.field_mapping_available) }, sort_keys=True)) return MetadataManifest( verdict=MetadataVerdict.CROSS_VERSION_TRANSLATED, directive=f"PROCEED: Translated {len(versions)} schema versions to {ctx.current_query_schema_version}.", manifest_digest=digest )