diff --git a/docs/vnext_schema_v1_foundation_contract.md b/docs/vnext_schema_v1_foundation_contract.md new file mode 100644 index 0000000..454a7e1 --- /dev/null +++ b/docs/vnext_schema_v1_foundation_contract.md @@ -0,0 +1,7 @@ +# QAR vNext schema-v1 identity foundation + +`qar_vnext_identity.v1` is a clean-slate immutable identity/index wire namespace. It is independent from legacy identity codecs and has no dual-read, migration, compatibility fallback, publisher, allocation, report verification, or I/O path. + +The top-level schema version and entry discriminator are fixed. The semantic and artifact algorithm pair is fixed for schema-v1; unknown/future/mismatched versions fail closed. Future algorithm changes require a new schema/namespace. + +The foundation validates period/cadence, report schema-contract pairing, digest/version fields, V3 canonical/variant class, pending status, cadence-aware targets, canonical/variant suffixes, optional attachment key presence, full artifact identity uniqueness, target/canonical/digest collisions, per-period display-primary/order invariants, safe integer bounds, and deterministic parse/serialize ordering. It does not claim that report bytes have been verified and does not create a publication plan. diff --git a/src/quant_advisor_research/vnext_identity_schema_v1.py b/src/quant_advisor_research/vnext_identity_schema_v1.py new file mode 100644 index 0000000..2674994 --- /dev/null +++ b/src/quant_advisor_research/vnext_identity_schema_v1.py @@ -0,0 +1,306 @@ +"""Pure clean-slate QAR vNext identity/index schema v1. + +This module deliberately stops at structural identity evidence. It has no +report verification, allocation, publication plan, legacy fallback, or I/O. +""" +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from .artifact_integrity import ArtifactIntegrityError, snapshot_json_wire +from .period_contract import PeriodContractError, canonical_period_identity +from .time_contract import TimeContractError, contract_version_for_schema + +VNEXT_SCHEMA_VERSION = 1 +VNEXT_NAMESPACE = "qar_vnext_identity.v1" +VNEXT_BINDING_NAMESPACE = "qar_vnext_binding.v1" +VNEXT_STATUS = "PENDING_ARTIFACT_VALIDATION" +SEMANTIC_ALGORITHM_VERSION = "semantic_fingerprint.v1.sha256" +ARTIFACT_ALGORITHM_VERSION = "validated_report.v1.canonical-json.sha256" +MAX_SAFE_JSON_INTEGER = 2**53 - 1 +V3_CANONICAL = "V3_CANONICAL" +V3_VARIANT = "V3_VARIANT" + +_DATE = r"(?P\d{4}-\d{2}-\d{2})" +_DIGEST = r"(?P[0-9a-f]{64})" +_PATTERNS = { + "json": re.compile(rf"^advisory_report_{_DATE}-(?Pdaily|weekly|monthly)(?:\.variant-{_DIGEST})?\.json$"), + "html": re.compile(rf"^{_DATE}-(?Pdaily|weekly|monthly)-model-recommendations(?:\.variant-{_DIGEST})?\.html$"), + "md": re.compile(rf"^advisory_report_{_DATE}-(?Pdaily|weekly|monthly)(?:\.variant-{_DIGEST})?\.md$"), + "manifest": re.compile(rf"^advisory_report_{_DATE}-(?Pdaily|weekly|monthly)(?:\.variant-{_DIGEST})?\.json\.manifest\.json$"), +} + + +class VNextIdentityError(ValueError): + """Stable, sanitized schema-v1 identity error.""" + + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +def _error(code: str) -> VNextIdentityError: + return VNextIdentityError(code) + + +def _exact(value: object, typ: type, code: str) -> Any: + if type(value) is not typ: + raise _error(code) + return value + + +def _digest(value: object, code: str) -> str: + value = _exact(value, str, code) + if re.fullmatch(r"[0-9a-f]{64}", value) is None: + raise _error(code) + return value + + +def _period_key(as_of: str, cadence: str) -> str: + try: + return canonical_period_identity(cadence, as_of).key + except (PeriodContractError, TypeError, ValueError, OverflowError): + raise _error("period_mismatch") from None + + +def _check_target(value: object, kind: str, *, as_of: str, cadence: str, suffix: str | None) -> str: + value = _exact(value, str, "target_invalid") + if not value or "/" in value or "\\" in value or value in {".", ".."}: + raise _error("target_invalid") + match = _PATTERNS[kind].fullmatch(value) + if match is None or match.group("as_of") != as_of or match.group("cadence") != cadence: + raise _error("target_mismatch") + if match.groupdict().get("digest") != suffix: + raise _error("target_digest_mismatch") + return value + + +@dataclass(frozen=True, slots=True) +class VNextIdentityBinding: + period_key: str + as_of: str + cadence: str + report_schema_version: str + contract_version: str + semantic_fingerprint_version: str + semantic_digest: str + artifact_integrity_version: str + artifact_integrity_digest: str + json_name: str + html_name: str + markdown_name: str | None + manifest_name: str | None + identity_class: str + canonical_identity: bool + display_primary: bool + display_order: int + status: str = VNEXT_STATUS + + def __post_init__(self) -> None: + _validate_binding(self) + + +@dataclass(frozen=True, slots=True) +class VNextIdentityIndex: + bindings: tuple[VNextIdentityBinding, ...] + schema_version: int = VNEXT_SCHEMA_VERSION + namespace: str = VNEXT_NAMESPACE + + def __post_init__(self) -> None: + if type(self.schema_version) is not int or self.schema_version != VNEXT_SCHEMA_VERSION: + raise _error("unsupported_schema") + if type(self.namespace) is not str or self.namespace != VNEXT_NAMESPACE: + raise _error("unsupported_namespace") + if type(self.bindings) is not tuple or not all(isinstance(item, VNextIdentityBinding) for item in self.bindings): + raise _error("index_invalid") + _validate_index(self.bindings) + object.__setattr__(self, "bindings", tuple(sorted(self.bindings, key=_binding_sort_key))) + + +def _validate_binding(binding: VNextIdentityBinding) -> None: + if not isinstance(binding, VNextIdentityBinding): + raise _error("binding_invalid") + text_fields = ( + binding.period_key, binding.as_of, binding.cadence, binding.report_schema_version, + binding.contract_version, binding.semantic_fingerprint_version, binding.semantic_digest, + binding.artifact_integrity_version, binding.artifact_integrity_digest, binding.json_name, + binding.html_name, binding.identity_class, binding.status, + ) + if any(type(value) is not str or not value for value in text_fields): + raise _error("binding_invalid") + if binding.period_key != _period_key(binding.as_of, binding.cadence): + raise _error("period_mismatch") + if binding.report_schema_version not in {"5", "6"}: + raise _error("report_schema_unsupported") + try: + expected = contract_version_for_schema(binding.report_schema_version) + except (TimeContractError, TypeError, ValueError): + raise _error("report_schema_unsupported") from None + if binding.contract_version != expected: + raise _error("contract_mismatch") + if binding.semantic_fingerprint_version != SEMANTIC_ALGORITHM_VERSION: + raise _error("semantic_algorithm_unsupported") + if binding.artifact_integrity_version != ARTIFACT_ALGORITHM_VERSION: + raise _error("artifact_algorithm_unsupported") + _digest(binding.semantic_digest, "semantic_digest_invalid") + _digest(binding.artifact_integrity_digest, "artifact_digest_invalid") + if binding.identity_class not in {V3_CANONICAL, V3_VARIANT}: + raise _error("identity_class_unsupported") + if type(binding.canonical_identity) is not bool or type(binding.display_primary) is not bool: + raise _error("boolean_invalid") + if binding.canonical_identity != (binding.identity_class == V3_CANONICAL): + raise _error("identity_class_mismatch") + if type(binding.display_order) is not int or type(binding.display_order) is bool or not 0 <= binding.display_order <= MAX_SAFE_JSON_INTEGER: + raise _error("display_order_invalid") + if binding.status != VNEXT_STATUS: + raise _error("status_invalid") + suffix = None if binding.identity_class == V3_CANONICAL else binding.artifact_integrity_digest + _check_target(binding.json_name, "json", as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + _check_target(binding.html_name, "html", as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + if binding.markdown_name is not None: + _check_target(binding.markdown_name, "md", as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + if binding.manifest_name is not None: + _check_target(binding.manifest_name, "manifest", as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + + +def _validate_index(bindings: tuple[VNextIdentityBinding, ...]) -> None: + canonical_periods: set[str] = set() + names: set[str] = set() + artifact_identities: set[tuple[str, str, str, str, str, str, str]] = set() + artifact_digests: set[str] = set() + display: dict[str, tuple[bool, set[int]]] = {} + for binding in bindings: + _validate_binding(binding) + exact = ( + binding.period_key, binding.as_of, binding.cadence, binding.report_schema_version, + binding.contract_version, binding.semantic_digest, binding.artifact_integrity_digest, + ) + if exact in artifact_identities: + raise _error("identity_duplicate") + artifact_identities.add(exact) + if binding.canonical_identity: + if binding.period_key in canonical_periods: + raise _error("canonical_conflict") + canonical_periods.add(binding.period_key) + primary, orders = display.setdefault(binding.period_key, (False, set())) + if binding.display_primary and primary: + raise _error("display_primary_conflict") + if binding.display_order in orders: + raise _error("display_order_conflict") + display[binding.period_key] = (primary or binding.display_primary, orders | {binding.display_order}) + for name in (binding.json_name, binding.html_name, binding.markdown_name, binding.manifest_name): + if name is not None: + if name in names: + raise _error("target_collision") + names.add(name) + if binding.artifact_integrity_digest in artifact_digests: + raise _error("artifact_digest_conflict") + artifact_digests.add(binding.artifact_integrity_digest) + if {item.period_key for item in bindings} - canonical_periods: + raise _error("canonical_missing") + + +def _snapshot(payload: Mapping[str, Any]) -> dict[str, object]: + try: + return snapshot_json_wire(payload) + except (ArtifactIntegrityError, TypeError, ValueError, UnicodeError, RecursionError): + raise _error("wire_invalid") from None + + +def _binding_sort_key(binding: VNextIdentityBinding) -> tuple[object, ...]: + return ( + binding.period_key, not binding.canonical_identity, binding.artifact_integrity_digest, + binding.semantic_digest, binding.json_name, binding.html_name, + binding.markdown_name or "", binding.manifest_name or "", + ) + + +def _from_entry(entry: object) -> VNextIdentityBinding: + required = { + "binding_namespace", "period_key", "as_of", "cadence", "report_schema_version", "contract_version", + "semantic_fingerprint_version", "semantic_digest", "artifact_integrity_version", "artifact_integrity_digest", + "json", "html", "identity_class", "canonical_identity", "display_primary", "display_order", "status", + } + optional = {"md", "manifest"} + if type(entry) is not dict or not required.issubset(entry) or set(entry) - required - optional: + raise _error("entry_invalid") + if entry["binding_namespace"] != VNEXT_BINDING_NAMESPACE: + raise _error("binding_namespace_unsupported") + if "md" in entry and type(entry["md"]) is not str: + raise _error("target_invalid") + if "manifest" in entry and type(entry["manifest"]) is not str: + raise _error("target_invalid") + try: + return VNextIdentityBinding( + period_key=entry["period_key"], as_of=entry["as_of"], cadence=entry["cadence"], + report_schema_version=entry["report_schema_version"], contract_version=entry["contract_version"], + semantic_fingerprint_version=entry["semantic_fingerprint_version"], semantic_digest=entry["semantic_digest"], + artifact_integrity_version=entry["artifact_integrity_version"], artifact_integrity_digest=entry["artifact_integrity_digest"], + json_name=entry["json"], html_name=entry["html"], markdown_name=entry.get("md"), manifest_name=entry.get("manifest"), + identity_class=entry["identity_class"], canonical_identity=entry["canonical_identity"], + display_primary=entry["display_primary"], display_order=entry["display_order"], status=entry["status"], + ) + except VNextIdentityError: + raise + except (AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("entry_invalid") from None + + +def parse_vnext_identity_index(payload: Mapping[str, Any]) -> VNextIdentityIndex: + snapshot = _snapshot(payload) + if set(snapshot) != {"schema_version", "namespace", "reports"}: + raise _error("wire_invalid") + if type(snapshot["schema_version"]) is not int or snapshot["schema_version"] != VNEXT_SCHEMA_VERSION: + raise _error("unsupported_schema") + if type(snapshot["namespace"]) is not str or snapshot["namespace"] != VNEXT_NAMESPACE: + raise _error("unsupported_namespace") + if type(snapshot["reports"]) is not list: + raise _error("wire_invalid") + try: + bindings = tuple(sorted((_from_entry(item) for item in snapshot["reports"]), key=_binding_sort_key)) + return VNextIdentityIndex(bindings) + except VNextIdentityError: + raise + except (AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("wire_invalid") from None + + +def _to_entry(binding: VNextIdentityBinding) -> dict[str, object]: + _validate_binding(binding) + entry: dict[str, object] = { + "binding_namespace": VNEXT_BINDING_NAMESPACE, "period_key": binding.period_key, "as_of": binding.as_of, + "cadence": binding.cadence, "report_schema_version": binding.report_schema_version, + "contract_version": binding.contract_version, "semantic_fingerprint_version": binding.semantic_fingerprint_version, + "semantic_digest": binding.semantic_digest, "artifact_integrity_version": binding.artifact_integrity_version, + "artifact_integrity_digest": binding.artifact_integrity_digest, "json": binding.json_name, + "html": binding.html_name, "identity_class": binding.identity_class, "canonical_identity": binding.canonical_identity, + "display_primary": binding.display_primary, "display_order": binding.display_order, "status": binding.status, + } + if binding.markdown_name is not None: + entry["md"] = binding.markdown_name + if binding.manifest_name is not None: + entry["manifest"] = binding.manifest_name + return entry + + +def serialize_vnext_identity_index(index: VNextIdentityIndex) -> dict[str, object]: + if not isinstance(index, VNextIdentityIndex): + raise _error("index_invalid") + _validate_index(index.bindings) + payload = { + "schema_version": VNEXT_SCHEMA_VERSION, "namespace": VNEXT_NAMESPACE, + "reports": [_to_entry(binding) for binding in sorted(index.bindings, key=_binding_sort_key)], + } + parse_vnext_identity_index(payload) + return payload + + +__all__ = [ + "ARTIFACT_ALGORITHM_VERSION", "MAX_SAFE_JSON_INTEGER", "SEMANTIC_ALGORITHM_VERSION", + "V3_CANONICAL", "V3_VARIANT", "VNEXT_BINDING_NAMESPACE", "VNEXT_NAMESPACE", "VNEXT_SCHEMA_VERSION", + "VNEXT_STATUS", "VNextIdentityBinding", "VNextIdentityError", "VNextIdentityIndex", + "parse_vnext_identity_index", "serialize_vnext_identity_index", +] diff --git a/tests/test_vnext_identity_schema_v1_foundation.py b/tests/test_vnext_identity_schema_v1_foundation.py new file mode 100644 index 0000000..7b6f1a5 --- /dev/null +++ b/tests/test_vnext_identity_schema_v1_foundation.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from collections import OrderedDict +from types import MappingProxyType + +import pytest + +from quant_advisor_research.vnext_identity_schema_v1 import ( + ARTIFACT_ALGORITHM_VERSION, + MAX_SAFE_JSON_INTEGER, + SEMANTIC_ALGORITHM_VERSION, + V3_CANONICAL, + V3_VARIANT, + VNEXT_BINDING_NAMESPACE, + VNEXT_NAMESPACE, + VNEXT_STATUS, + VNextIdentityBinding, + VNextIdentityError, + VNextIdentityIndex, + parse_vnext_identity_index, + serialize_vnext_identity_index, +) +from quant_advisor_research.period_contract import canonical_period_identity + +A = "a" * 64 +B = "b" * 64 + + +def binding(*, as_of="2026-06-20", cadence="weekly", cls=V3_CANONICAL, artifact=A, + semantic=A, md=False, manifest=False, primary=False, order=0): + suffix = "" if cls == V3_CANONICAL else f".variant-{artifact}" + stem = f"advisory_report_{as_of}-{cadence}{suffix}" + return VNextIdentityBinding( + period_key=canonical_period_identity(cadence, as_of).key, + as_of=as_of, cadence=cadence, report_schema_version="5", contract_version="model_recommendations.v5", + semantic_fingerprint_version=SEMANTIC_ALGORITHM_VERSION, semantic_digest=semantic, + artifact_integrity_version=ARTIFACT_ALGORITHM_VERSION, artifact_integrity_digest=artifact, + json_name=f"{stem}.json", html_name=f"{as_of}-{cadence}-model-recommendations{suffix}.html", + markdown_name=f"{stem}.md" if md else None, + manifest_name=f"{stem}.json.manifest.json" if manifest else None, + identity_class=cls, canonical_identity=cls == V3_CANONICAL, + display_primary=primary, display_order=order, status=VNEXT_STATUS, + ) + + +def payload(*bindings): + return {"schema_version": 1, "namespace": VNEXT_NAMESPACE, + "reports": [wire_entry(item) for item in bindings]} + + +def wire_entry(item): + value = { + "binding_namespace": VNEXT_BINDING_NAMESPACE, "period_key": item.period_key, "as_of": item.as_of, + "cadence": item.cadence, "report_schema_version": item.report_schema_version, + "contract_version": item.contract_version, "semantic_fingerprint_version": item.semantic_fingerprint_version, + "semantic_digest": item.semantic_digest, "artifact_integrity_version": item.artifact_integrity_version, + "artifact_integrity_digest": item.artifact_integrity_digest, "json": item.json_name, + "html": item.html_name, "identity_class": item.identity_class, "canonical_identity": item.canonical_identity, + "display_primary": item.display_primary, "display_order": item.display_order, "status": item.status, + } + if item.markdown_name is not None: + value["md"] = item.markdown_name + if item.manifest_name is not None: + value["manifest"] = item.manifest_name + return value + + +def test_schema_v1_roundtrip_and_input_permutation_are_deterministic(): + canonical = binding() + variant = binding(cls=V3_VARIANT, artifact=B, semantic=B, order=1, md=True, manifest=True) + left = parse_vnext_identity_index(payload(canonical, variant)) + right = parse_vnext_identity_index(payload(variant, canonical)) + assert left == right + assert hash(left) == hash(right) + assert serialize_vnext_identity_index(left) == serialize_vnext_identity_index(right) + assert parse_vnext_identity_index(serialize_vnext_identity_index(left)) == left + + +def test_cadence_aware_targets_allow_same_date_daily_weekly_monthly(): + entries = [binding(cadence=cadence, artifact=digest, semantic=digest, order=position) + for position, (cadence, digest) in enumerate((("daily", A), ("weekly", B), ("monthly", "c" * 64)))] + index = VNextIdentityIndex(tuple(entries)) + assert len({item.json_name for item in index.bindings}) == 3 + + +@pytest.mark.parametrize("field,value", [ + ("schema_version", 2), ("namespace", "legacy"), ("binding_namespace", "legacy"), + ("report_schema_version", 6), ("contract_version", "model_recommendations.v6"), + ("semantic_fingerprint_version", "future"), ("artifact_integrity_version", "future"), + ("identity_class", "LEGACY_V2"), ("status", "VERIFIED"), +]) +def test_unknown_future_legacy_and_mismatched_versions_fail_closed(field, value): + item = binding() + wire = payload(item) + if field in {"schema_version", "namespace"}: + wire[field] = value + else: + wire["reports"][0][field] = value + with pytest.raises(VNextIdentityError): + parse_vnext_identity_index(wire) + + +@pytest.mark.parametrize("optional", ["md", "manifest"]) +def test_optional_attachment_omission_is_distinct_from_null(optional): + item = binding() + wire = payload(item) + wire["reports"][0][optional] = None + with pytest.raises(VNextIdentityError, match="target_invalid"): + parse_vnext_identity_index(wire) + assert parse_vnext_identity_index(payload(item)).bindings[0].markdown_name is None + + +def test_variant_all_declared_targets_use_full_artifact_digest(): + item = binding(cls=V3_VARIANT, artifact=B, semantic=A, md=True, manifest=True, order=1) + wire = payload(binding(), item) + assert len(parse_vnext_identity_index(wire).bindings) == 2 + wire["reports"][1]["md"] = wire["reports"][1]["md"].replace(B, A) + with pytest.raises(VNextIdentityError, match="target_digest_mismatch"): + parse_vnext_identity_index(wire) + + +def test_index_requires_one_canonical_and_enforces_display_policy(): + variant = binding(cls=V3_VARIANT, artifact=B, order=1) + with pytest.raises(VNextIdentityError, match="canonical_missing"): + VNextIdentityIndex((variant,)) + second = binding(cls=V3_VARIANT, artifact=B, order=0) + with pytest.raises(VNextIdentityError, match="display_order_conflict"): + VNextIdentityIndex((binding(), second)) + with pytest.raises(VNextIdentityError, match="display_primary_conflict"): + VNextIdentityIndex((binding(primary=True), binding(cls=V3_VARIANT, artifact=B, primary=True, order=1))) + + +def test_full_artifact_identity_and_target_collisions_fail_closed(): + with pytest.raises(VNextIdentityError, match="identity_duplicate"): + VNextIdentityIndex((binding(), binding(md=True))) + with pytest.raises(VNextIdentityError, match="artifact_digest_conflict"): + VNextIdentityIndex((binding(), binding(cls=V3_VARIANT, semantic=B, order=1))) + + +def test_display_order_is_safe_json_integer_and_not_bool(): + assert binding(order=MAX_SAFE_JSON_INTEGER).display_order == MAX_SAFE_JSON_INTEGER + for value in (-1, MAX_SAFE_JSON_INTEGER + 1, True, "0"): + with pytest.raises(VNextIdentityError, match="display_order_invalid"): + binding(order=value) + + +def test_mapping_snapshot_and_wire_shape_are_sanitized(): + item = binding() + value = payload(item) + ordered = OrderedDict((key, value[key]) for key in reversed(value)) + assert parse_vnext_identity_index(MappingProxyType(ordered)) == parse_vnext_identity_index(value) + with pytest.raises(VNextIdentityError, match="wire_invalid"): + parse_vnext_identity_index({"schema_version": 1, "namespace": VNEXT_NAMESPACE, "reports": [{"raw": object()}]})