From 7909f94dacc86a7d2b99abed2c20cf65e6facb45 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:34:37 +0800 Subject: [PATCH 1/2] feat: add clean-slate vnext schema v1 foundation Co-Authored-By: Codex --- docs/vnext_schema_v1_contract.md | 23 ++ .../vnext_identity_v1.py | 336 ++++++++++++++++++ .../vnext_publication_plan.py | 306 ++++++++++++++++ tests/test_vnext_schema_v1.py | 253 +++++++++++++ 4 files changed, 918 insertions(+) create mode 100644 docs/vnext_schema_v1_contract.md create mode 100644 src/quant_advisor_research/vnext_identity_v1.py create mode 100644 src/quant_advisor_research/vnext_publication_plan.py create mode 100644 tests/test_vnext_schema_v1.py diff --git a/docs/vnext_schema_v1_contract.md b/docs/vnext_schema_v1_contract.md new file mode 100644 index 0000000..c009897 --- /dev/null +++ b/docs/vnext_schema_v1_contract.md @@ -0,0 +1,23 @@ +# QAR clean-slate vNext identity schema v1 + +本契约是独立的 clean-slate namespace：`qar_vnext_identity.v1`。它不读取、迁移、猜测或回退到 legacy index、legacy filename、旧 publisher/recovery 输出；legacy 类型和旧 schema 不进入本路径。 + +## 固定版本 + +- 顶层 `schema_version` 固定为整数 `1`。 +- 顶层 `namespace` 固定为 `qar_vnext_identity.v1`；entry 的 `binding_namespace` 固定为 `qar_vnext_binding.v1`。 +- `semantic_fingerprint_version` 固定为 `semantic_fingerprint.v1.sha256`。 +- `artifact_integrity_version` 固定为 `validated_report.v1.canonical-json.sha256`。 +- 只接受当前 exact pair。未知、future、mismatch 或运行时可变版本一律 fail-closed；未来算法变化创建独立 schema/namespace，不做 registry、dual-read 或 migration。 + +## Entry 与 target + +`report_schema_version` 必须是字符串 `5` 或 `6`，`contract_version` 必须由唯一 `contract_version_for_schema()` 严格匹配。所有 target 都包含 `as_of + cadence`：canonical 无 suffix，variant 的 json/html/md/manifest（只校验实际声明者）都使用完整 artifact digest suffix。md/manifest key omission 表示未请求；key present 必须是有效 basename string，显式 `null` 不等同 omission。 + +完整 index 每个 period 恰好一个 canonical；variant-only、多个 canonical、basename/digest collision、错误 period/schema/contract/class/status、legacy target 均拒绝。`display_primary` 每个 period 最多一个，`display_order` 每个 period 唯一且是 `0..2**53-1` 的非 bool integer。identity owner 与 display placement 分离，后续 PublicationPlan 可使用独立 display evidence。 + +## Pure boundary + +`VNextIdentityBinding`、`VNextIdentityIndex`、allocation modes 和 `PublicationPlan` 均为 immutable/pure value objects。raw report、index、attachments、display placement 先校验；source basename 不决定 public target。CURRENT 空 period 才能 bootstrap canonical；occupied period 的 changed artifact 分配 variant；identical artifact 复用；HISTORICAL 无 canonical fail-closed；EXACT 只复用 exact。 + +本阶段不包含 filesystem/network/publisher/build/workflow/Pages/I/O、legacy compatibility/migration、真实发布或 N2。 diff --git a/src/quant_advisor_research/vnext_identity_v1.py b/src/quant_advisor_research/vnext_identity_v1.py new file mode 100644 index 0000000..f49bed0 --- /dev/null +++ b/src/quant_advisor_research/vnext_identity_v1.py @@ -0,0 +1,336 @@ +"""Clean-slate QAR vNext identity-index schema v1. + +This namespace is intentionally independent from legacy identity codecs. It +accepts exactly one immutable algorithm pair; future algorithm changes require +another schema/namespace rather than read compatibility or fallback. +""" +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from .artifact_integrity import ArtifactIntegrityError, ARTIFACT_INTEGRITY_VERSION, make_artifact_integrity_evidence, snapshot_json_wire +from .contracts import AdvisoryValidationError, validate_advisory_report +from .period_contract import PeriodContractError, canonical_period_identity +from .time_contract import TimeContractError, contract_version_for_schema + +VNEXT_SCHEMA_VERSION = 1 +VNEXT_WIRE_NAMESPACE = "qar_vnext_identity.v1" +VNEXT_BINDING_NAMESPACE = "qar_vnext_binding.v1" +VNEXT_STATUS = "PENDING_ARTIFACT_VALIDATION" +SEMANTIC_FINGERPRINT_VERSION = "semantic_fingerprint.v1.sha256" +ARTIFACT_ALGORITHM_VERSION = ARTIFACT_INTEGRITY_VERSION +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})" +_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): + 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(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 _name(value: object, pattern: re.Pattern[str], *, as_of: str, cadence: str, suffix: str | None) -> str: + value = _exact(value, str, "invalid_target_name") + if not value or "/" in value or "\\" in value or value in {".", ".."}: + raise _error("invalid_target_name") + match = pattern.fullmatch(value) + if match is None or match.group("as_of") != as_of or match.group("cadence") != cadence: + raise _error("target_name_mismatch") + if match.groupdict().get("digest") != suffix: + raise _error("target_digest_mismatch") + return value + + +def _semantic_digest(snapshot: Mapping[str, Any]) -> str: + ignored = {"as_of", "generated_at", "reference_time", "expires_at", "schema_version", "contract_version", "source_artifacts"} + normalized = {key: value for key, value in snapshot.items() if key not in ignored} + freshness = normalized.get("freshness") + if isinstance(freshness, dict): + normalized["freshness"] = { + name: {key: entry[key] for key in ("present", "valid", "reason") if key in entry} + for name, entry in freshness.items() if isinstance(entry, dict) + } + elif "freshness" not in normalized: + normalized["freshness"] = { + "ai_signal": {"present": False, "valid": False, "reason": "not_provided"}, + "theme_momentum": {"present": False, "valid": False, "reason": "not_provided"}, + } + summary = normalized.get("summary") + if isinstance(summary, dict) and isinstance(summary.get("data_quality_warnings"), list): + summary = dict(summary) + summary["data_quality_warnings"] = [ + item for item in summary["data_quality_warnings"] + if not (isinstance(item, str) and item.startswith(("compatibility:", "compatibility_", "schema_compatibility:", "schema_compatibility_"))) + ] + normalized["summary"] = summary + try: + return hashlib.sha256(json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("report_invalid") from None + + +@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_WIRE_NAMESPACE + + def __post_init__(self) -> None: + if type(self.schema_version) is not int or self.schema_version != VNEXT_SCHEMA_VERSION or self.namespace != VNEXT_WIRE_NAMESPACE: + raise _error("unsupported_index_version") + if type(self.bindings) is not tuple or not all(isinstance(item, VNextIdentityBinding) for item in self.bindings): + raise _error("identity_index_invalid") + _validate_index(self.bindings) + + +def _validate_binding(binding: VNextIdentityBinding) -> None: + if not isinstance(binding, VNextIdentityBinding): + raise _error("identity_binding_invalid") + for value in (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 type(value) is not str or not value: + raise _error("identity_binding_invalid") + expected_period = _period(binding.as_of, binding.cadence) + if binding.period_key != expected_period: + raise _error("period_mismatch") + if binding.report_schema_version not in {"5", "6"}: + raise _error("unsupported_report_schema") + try: + expected_contract = contract_version_for_schema(binding.report_schema_version) + except (TimeContractError, TypeError, ValueError): + raise _error("unsupported_report_schema") from None + if binding.contract_version != expected_contract: + raise _error("contract_version_mismatch") + if binding.semantic_fingerprint_version != SEMANTIC_FINGERPRINT_VERSION: + raise _error("unsupported_semantic_algorithm") + if binding.artifact_integrity_version != ARTIFACT_ALGORITHM_VERSION: + raise _error("unsupported_artifact_algorithm") + _digest(binding.semantic_digest, "invalid_semantic_digest") + _digest(binding.artifact_integrity_digest, "invalid_artifact_digest") + if binding.identity_class not in {V3_CANONICAL, V3_VARIANT}: + raise _error("unsupported_identity_class") + if type(binding.canonical_identity) is not bool or type(binding.display_primary) is not bool: + raise _error("identity_boolean_invalid") + if (binding.identity_class == V3_CANONICAL) != binding.canonical_identity: + 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("identity_status_invalid") + suffix = None if binding.identity_class == V3_CANONICAL else binding.artifact_integrity_digest + _name(binding.json_name, _JSON, as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + _name(binding.html_name, _HTML, as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + if binding.markdown_name is not None: + _name(binding.markdown_name, _MD, as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + if binding.manifest_name is not None: + _name(binding.manifest_name, _MANIFEST, as_of=binding.as_of, cadence=binding.cadence, suffix=suffix) + + +def _validate_index(bindings: tuple[VNextIdentityBinding, ...]) -> None: + seen_names: dict[str, tuple[str, str]] = {} + seen_identities: set[tuple[str, str, str, str | None, str | None]] = set() + canonical_by_period: set[str] = set() + display_by_period: dict[str, tuple[bool, set[int]]] = {} + artifact_digests: dict[str, tuple[str, str, str]] = {} + for binding in bindings: + _validate_binding(binding) + identity = (binding.period_key, binding.json_name, binding.html_name, binding.markdown_name, binding.manifest_name) + if binding.canonical_identity: + if binding.period_key in canonical_by_period: + raise _error("canonical_conflict") + canonical_by_period.add(binding.period_key) + if identity in seen_identities: + raise _error("identity_duplicate") + seen_identities.add(identity) + primary, orders = display_by_period.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_by_period[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 None: + continue + owner = (binding.period_key, binding.artifact_integrity_digest) + if name in seen_names and seen_names[name] != owner: + raise _error("target_collision") + seen_names[name] = owner + previous = artifact_digests.get(binding.artifact_integrity_digest) + current = (binding.period_key, binding.as_of, binding.cadence) + if previous is not None and previous != current: + raise _error("artifact_digest_conflict") + artifact_digests[binding.artifact_integrity_digest] = current + periods = {binding.period_key for binding in bindings} + if periods - canonical_by_period: + 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("invalid_wire") from None + + +def _entry_from_wire(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("invalid_entry") + if entry.get("binding_namespace") != VNEXT_BINDING_NAMESPACE: + raise _error("unsupported_binding_namespace") + try: + if "md" in entry and type(entry["md"]) is not str: + raise _error("invalid_target_name") + if "manifest" in entry and type(entry["manifest"]) is not str: + raise _error("invalid_target_name") + return VNextIdentityBinding( + entry["period_key"], entry["as_of"], entry["cadence"], entry["report_schema_version"], entry["contract_version"], + entry["semantic_fingerprint_version"], entry["semantic_digest"], entry["artifact_integrity_version"], + entry["artifact_integrity_digest"], entry["json"], entry["html"], entry.get("md"), entry.get("manifest"), + entry["identity_class"], entry["canonical_identity"], entry["display_primary"], entry["display_order"], entry["status"], + ) + except VNextIdentityError: + raise + except (AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("invalid_entry") from None + + +def parse_vnext_index(payload: Mapping[str, Any]) -> VNextIdentityIndex: + snapshot = _snapshot(payload) + if set(snapshot) != {"schema_version", "namespace", "reports"}: + raise _error("invalid_wire") + if type(snapshot["schema_version"]) is not int or snapshot["schema_version"] != VNEXT_SCHEMA_VERSION: + raise _error("unsupported_index_version") + if type(snapshot["namespace"]) is not str or snapshot["namespace"] != VNEXT_WIRE_NAMESPACE: + raise _error("unsupported_namespace") + if type(snapshot["reports"]) is not list: + raise _error("invalid_wire") + try: + bindings = tuple(_entry_from_wire(item) for item in snapshot["reports"]) + return VNextIdentityIndex(bindings) + except VNextIdentityError: + raise + except (AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("invalid_wire") from None + + +def _binding_wire(binding: VNextIdentityBinding) -> dict[str, object]: + _validate_binding(binding) + result = { + "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: + result["md"] = binding.markdown_name + if binding.manifest_name is not None: + result["manifest"] = binding.manifest_name + return result + + +def serialize_vnext_index(index: VNextIdentityIndex) -> dict[str, object]: + if not isinstance(index, VNextIdentityIndex): + raise _error("identity_index_invalid") + _validate_index(index.bindings) + payload = {"schema_version": VNEXT_SCHEMA_VERSION, "namespace": VNEXT_WIRE_NAMESPACE, + "reports": [_binding_wire(binding) for binding in index.bindings]} + parse_vnext_index(payload) + return payload + + +def report_identity_evidence(report: Mapping[str, Any]) -> tuple[dict[str, object], str, str, str, str, str, str, str]: + try: + snapshot = snapshot_json_wire(report) + validate_advisory_report(snapshot) + schema = snapshot["schema_version"] + as_of = snapshot["as_of"] + cadence = snapshot["cadence"] + if not all(type(value) is str for value in (schema, as_of, cadence)): + raise _error("report_invalid") + contract = contract_version_for_schema(schema) + period = _period(as_of, cadence) + semantic = _semantic_digest(snapshot) + artifact = make_artifact_integrity_evidence(snapshot) + return snapshot, period, as_of, cadence, schema, contract, semantic, artifact.digest + except VNextIdentityError: + raise + except (AdvisoryValidationError, ArtifactIntegrityError, PeriodContractError, TimeContractError, + AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("report_invalid") from None + + +__all__ = [ + "ARTIFACT_ALGORITHM_VERSION", "MAX_SAFE_JSON_INTEGER", "SEMANTIC_FINGERPRINT_VERSION", "V3_CANONICAL", "V3_VARIANT", + "VNEXT_BINDING_NAMESPACE", "VNEXT_SCHEMA_VERSION", "VNEXT_STATUS", "VNEXT_WIRE_NAMESPACE", "VNextIdentityBinding", + "VNextIdentityError", "VNextIdentityIndex", "parse_vnext_index", "report_identity_evidence", "serialize_vnext_index", +] diff --git a/src/quant_advisor_research/vnext_publication_plan.py b/src/quant_advisor_research/vnext_publication_plan.py new file mode 100644 index 0000000..12f5b41 --- /dev/null +++ b/src/quant_advisor_research/vnext_publication_plan.py @@ -0,0 +1,306 @@ +"""Pure clean-slate vNext allocation and publication-plan contract. + +No filesystem, network, publisher, or legacy compatibility is involved. A +source report is evidence; its basename never chooses a public target. +""" +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from enum import Enum +from typing import Any + +from .vnext_identity_v1 import ( + MAX_SAFE_JSON_INTEGER, + SEMANTIC_FINGERPRINT_VERSION, + ARTIFACT_ALGORITHM_VERSION, + V3_CANONICAL, + V3_VARIANT, + VNEXT_STATUS, + VNextIdentityBinding, + VNextIdentityError, + VNextIdentityIndex, + report_identity_evidence, +) + + +class VNextPublicationError(ValueError): + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +class AllocationMode(Enum): + EXACT_ARTIFACT_REUSE = "EXACT_ARTIFACT_REUSE" + CURRENT_MANDATORY = "CURRENT_MANDATORY" + HISTORICAL_RECOVERY = "HISTORICAL_RECOVERY" + + +class PublicationRole(Enum): + MANDATORY_CURRENT = "MANDATORY_CURRENT" + RECOVERED_HISTORY = "RECOVERED_HISTORY" + + +def _error(code: str) -> VNextPublicationError: + return VNextPublicationError(code) + + +def _source(value: object) -> str: + if type(value) is not str or not value or value.strip() != value or "/" in value or "\\" in value or value in {".", ".."}: + raise _error("source_identity_invalid") + return value + + +def _display(primary: object, order: object) -> tuple[bool, int]: + if type(primary) is not bool or type(order) is not int or type(order) is bool or not 0 <= order <= MAX_SAFE_JSON_INTEGER: + raise _error("display_invalid") + return primary, order + + +def _freeze(value: object) -> object: + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze(item) for key, item in value.items()}) + if isinstance(value, list): + return tuple(_freeze(item) for item in value) + return value + + +@dataclass(frozen=True, slots=True) +class RequestedArtifacts: + markdown: bool = False + manifest: bool = False + + def __post_init__(self) -> None: + if type(self.markdown) is not bool or type(self.manifest) is not bool: + raise _error("attachment_policy_invalid") + + +@dataclass(frozen=True, slots=True) +class DisplayPlacement: + primary: bool + order: int + + def __post_init__(self) -> None: + _display(self.primary, self.order) + + +@dataclass(frozen=True, slots=True) +class AllocationContext: + mode: AllocationMode + requested_artifacts: RequestedArtifacts + display: DisplayPlacement + target_period_key: str | None = None + + def __post_init__(self) -> None: + if type(self.mode) is not AllocationMode: + raise _error("allocation_context_invalid") + if not isinstance(self.requested_artifacts, RequestedArtifacts) or not isinstance(self.display, DisplayPlacement): + raise _error("allocation_context_invalid") + if self.target_period_key is not None and type(self.target_period_key) is not str: + raise _error("allocation_context_invalid") + if self.mode is AllocationMode.CURRENT_MANDATORY and not self.target_period_key: + raise _error("allocation_context_required") + if self.mode is not AllocationMode.CURRENT_MANDATORY and self.target_period_key is not None: + raise _error("allocation_context_invalid") + + +@dataclass(frozen=True, slots=True) +class SelectedCandidate: + report_snapshot: Mapping[str, object] + source_identity: str + period_key: str + as_of: str + cadence: str + report_schema_version: str + contract_version: str + semantic_digest: str + artifact_integrity_digest: str + + @classmethod + def from_report(cls, report: Mapping[str, Any], *, source_identity: str) -> "SelectedCandidate": + _source(source_identity) + try: + snapshot, period, as_of, cadence, schema, contract, semantic, artifact = report_identity_evidence(report) + return cls(MappingProxyType({key: _freeze(value) for key, value in snapshot.items()}), source_identity, + period, as_of, cadence, schema, contract, semantic, artifact) + except VNextPublicationError: + raise + except (VNextIdentityError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("candidate_invalid") from None + + +@dataclass(frozen=True, slots=True) +class PublicationEntry: + candidate: SelectedCandidate + binding: VNextIdentityBinding + role: PublicationRole + display_primary: bool + display_order: int + + def __post_init__(self) -> None: + if not isinstance(self.candidate, SelectedCandidate) or not isinstance(self.binding, VNextIdentityBinding): + raise _error("publication_entry_invalid") + if type(self.role) is not PublicationRole: + raise _error("publication_role_invalid") + _display(self.display_primary, self.display_order) + _validate_candidate_binding(self.candidate, self.binding) + + +@dataclass(frozen=True, slots=True) +class PublicationPlan: + entries: tuple[PublicationEntry, ...] + + def __post_init__(self) -> None: + if type(self.entries) is not tuple or not self.entries: + raise _error("publication_plan_invalid") + _validate_plan(self.entries) + + +@dataclass(frozen=True, slots=True) +class AllocationResult: + binding: VNextIdentityBinding + reused_existing: bool + + +def _validate_candidate_binding(candidate: SelectedCandidate, binding: VNextIdentityBinding) -> None: + if (candidate.period_key, candidate.as_of, candidate.cadence, candidate.report_schema_version, candidate.contract_version, + candidate.semantic_digest, candidate.artifact_integrity_digest) != ( + binding.period_key, binding.as_of, binding.cadence, binding.report_schema_version, binding.contract_version, + binding.semantic_digest, binding.artifact_integrity_digest): + raise _error("candidate_binding_mismatch") + + +def _names(as_of: str, cadence: str, *, variant_digest: str | None, artifacts: RequestedArtifacts) -> tuple[str, str, str | None, str | None]: + suffix = "" if variant_digest is None else f".variant-{variant_digest}" + stem = f"advisory_report_{as_of}-{cadence}{suffix}" + return ( + f"{stem}.json", + f"{as_of}-{cadence}-model-recommendations{suffix}.html", + f"{stem}.md" if artifacts.markdown else None, + f"{stem}.json.manifest.json" if artifacts.manifest else None, + ) + + +def _new_binding(candidate: SelectedCandidate, identity_class: str, artifacts: RequestedArtifacts, display: DisplayPlacement) -> VNextIdentityBinding: + if identity_class not in {V3_CANONICAL, V3_VARIANT}: + raise _error("unsupported_identity_class") + json_name, html_name, markdown_name, manifest_name = _names( + candidate.as_of, candidate.cadence, + variant_digest=None if identity_class == V3_CANONICAL else candidate.artifact_integrity_digest, + artifacts=artifacts, + ) + return VNextIdentityBinding( + candidate.period_key, candidate.as_of, candidate.cadence, candidate.report_schema_version, + candidate.contract_version, SEMANTIC_FINGERPRINT_VERSION, candidate.semantic_digest, + ARTIFACT_ALGORITHM_VERSION, candidate.artifact_integrity_digest, json_name, html_name, + markdown_name, manifest_name, identity_class, identity_class == V3_CANONICAL, + display.primary, display.order, VNEXT_STATUS, + ) + + +def _policy_matches(binding: VNextIdentityBinding, artifacts: RequestedArtifacts) -> bool: + return (binding.markdown_name is not None) == artifacts.markdown and (binding.manifest_name is not None) == artifacts.manifest + + +def _candidate_key(candidate: SelectedCandidate) -> tuple[object, ...]: + return (candidate.period_key, candidate.as_of, candidate.cadence, candidate.report_schema_version, + candidate.contract_version, candidate.semantic_digest, candidate.artifact_integrity_digest) + + +def _same_artifact(a: SelectedCandidate, b: VNextIdentityBinding) -> bool: + return _candidate_key(a) == (b.period_key, b.as_of, b.cadence, b.report_schema_version, b.contract_version, + b.semantic_digest, b.artifact_integrity_digest) + + +def _simulate(index: VNextIdentityIndex, candidate: SelectedCandidate, binding: VNextIdentityBinding) -> None: + try: + _validate_candidate_binding(candidate, binding) + VNextIdentityIndex(index.bindings + (binding,)) + except VNextPublicationError: + raise + except VNextIdentityError as exc: + raise _error(exc.code) from None + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + raise _error("identity_simulation_invalid") from None + + +def allocate_identity(index: VNextIdentityIndex, candidate: SelectedCandidate, context: AllocationContext) -> AllocationResult: + if not isinstance(index, VNextIdentityIndex) or not isinstance(candidate, SelectedCandidate) or not isinstance(context, AllocationContext): + raise _error("allocation_input_invalid") + try: + VNextIdentityIndex(index.bindings) + except VNextIdentityError as exc: + raise _error(exc.code) from None + if context.mode is AllocationMode.CURRENT_MANDATORY and context.target_period_key != candidate.period_key: + raise _error("allocation_context_mismatch") + same = [binding for binding in index.bindings if _same_artifact(candidate, binding)] + if len(same) > 1: + raise _error("identity_conflict") + if same: + binding = same[0] + if not _policy_matches(binding, context.requested_artifacts): + raise _error("identity_reuse_mismatch") + if context.mode is AllocationMode.EXACT_ARTIFACT_REUSE and ( + binding.display_primary != context.display.primary or binding.display_order != context.display.order + ): + raise _error("identity_reuse_mismatch") + if context.mode is AllocationMode.EXACT_ARTIFACT_REUSE: + return AllocationResult(binding, True) + return AllocationResult(binding, True) + if context.mode is AllocationMode.EXACT_ARTIFACT_REUSE: + raise _error("identity_reuse_not_found") + canonical = [binding for binding in index.bindings if binding.period_key == candidate.period_key and binding.canonical_identity] + if context.mode is AllocationMode.HISTORICAL_RECOVERY and not canonical: + raise _error("canonical_bootstrap_required") + identity_class = V3_VARIANT if canonical else V3_CANONICAL + next_order = max((item.display_order for item in index.bindings if item.period_key == candidate.period_key), default=-1) + 1 + binding = _new_binding( + candidate, + identity_class, + context.requested_artifacts, + DisplayPlacement(False, next_order), + ) + _simulate(index, candidate, binding) + return AllocationResult(binding, False) + + +def _entry_key(entry: PublicationEntry) -> tuple[object, ...]: + return (entry.candidate.period_key, entry.display_order, not entry.display_primary, entry.role.value, entry.binding.json_name) + + +def _validate_plan(entries: tuple[PublicationEntry, ...]) -> None: + if sum(entry.role is PublicationRole.MANDATORY_CURRENT for entry in entries) != 1: + raise _error("mandatory_current_invalid") + target_seen: set[str] = set() + display: dict[str, tuple[bool, set[int]]] = {} + for entry in entries: + if entry.role is PublicationRole.MANDATORY_CURRENT and entry.binding.period_key != entry.candidate.period_key: + raise _error("mandatory_current_invalid") + for name in (entry.binding.json_name, entry.binding.html_name, entry.binding.markdown_name, entry.binding.manifest_name): + if name is not None and name in target_seen: + raise _error("publication_target_collision") + if name is not None: + target_seen.add(name) + primary, orders = display.setdefault(entry.binding.period_key, (False, set())) + if entry.display_primary and primary: + raise _error("display_primary_conflict") + if entry.display_order in orders: + raise _error("display_order_conflict") + display[entry.binding.period_key] = (primary or entry.display_primary, orders | {entry.display_order}) + + +def build_publication_plan(entries: list[PublicationEntry] | tuple[PublicationEntry, ...]) -> PublicationPlan: + if type(entries) not in {list, tuple}: + raise _error("publication_plan_invalid") + for entry in entries: + if not isinstance(entry, PublicationEntry): + raise _error("publication_entry_invalid") + return PublicationPlan(tuple(sorted(entries, key=_entry_key))) + + +__all__ = [ + "AllocationContext", "AllocationMode", "AllocationResult", "DisplayPlacement", "PublicationEntry", "PublicationPlan", + "PublicationRole", "RequestedArtifacts", "SelectedCandidate", "VNextPublicationError", "allocate_identity", + "build_publication_plan", +] diff --git a/tests/test_vnext_schema_v1.py b/tests/test_vnext_schema_v1.py new file mode 100644 index 0000000..8698776 --- /dev/null +++ b/tests/test_vnext_schema_v1.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +from collections import OrderedDict +import json +from pathlib import Path +from types import MappingProxyType + +import pytest + +from quant_advisor_research.advisory_report import build_advisory_report +from quant_advisor_research.vnext_identity_v1 import ( + ARTIFACT_ALGORITHM_VERSION, + MAX_SAFE_JSON_INTEGER, + SEMANTIC_FINGERPRINT_VERSION, + V3_CANONICAL, + V3_VARIANT, + VNEXT_BINDING_NAMESPACE, + VNEXT_STATUS, + VNextIdentityBinding, + VNextIdentityError, + VNextIdentityIndex, + parse_vnext_index, + serialize_vnext_index, +) +from quant_advisor_research.vnext_publication_plan import ( + AllocationContext, + AllocationMode, + DisplayPlacement, + PublicationEntry, + PublicationRole, + RequestedArtifacts, + SelectedCandidate, + VNextPublicationError, + allocate_identity, + build_publication_plan, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def report(*, as_of="2026-06-20", cadence="weekly", generated_at=None): + value = build_advisory_report( + as_of=as_of, cadence=cadence, + political_events_path=ROOT / "examples/political_events.example.csv", + political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", + ) + if generated_at is not None: + value["generated_at"] = generated_at + return value + + +def candidate(value, source="source"): + return SelectedCandidate.from_report(value, source_identity=source) + + +def binding(c, cls=V3_CANONICAL, *, md=False, manifest=False, primary=False, order=0): + suffix = "" if cls == V3_CANONICAL else f".variant-{c.artifact_integrity_digest}" + stem = f"advisory_report_{c.as_of}-{c.cadence}{suffix}" + return VNextIdentityBinding( + c.period_key, c.as_of, c.cadence, c.report_schema_version, c.contract_version, + SEMANTIC_FINGERPRINT_VERSION, c.semantic_digest, ARTIFACT_ALGORITHM_VERSION, + c.artifact_integrity_digest, f"{stem}.json", + f"{c.as_of}-{c.cadence}-model-recommendations{suffix}.html", + f"{stem}.md" if md else None, + f"{stem}.json.manifest.json" if manifest else None, + cls, cls == V3_CANONICAL, primary, order, VNEXT_STATUS, + ) + + +def wire(*entries): + return {"schema_version": 1, "namespace": "qar_vnext_identity.v1", "reports": list(entries)} + + +def entry(c, cls=V3_CANONICAL, **kwargs): + b = binding(c, cls, **kwargs) + result = { + "binding_namespace": VNEXT_BINDING_NAMESPACE, + "period_key": b.period_key, "as_of": b.as_of, "cadence": b.cadence, + "report_schema_version": b.report_schema_version, "contract_version": b.contract_version, + "semantic_fingerprint_version": b.semantic_fingerprint_version, "semantic_digest": b.semantic_digest, + "artifact_integrity_version": b.artifact_integrity_version, "artifact_integrity_digest": b.artifact_integrity_digest, + "json": b.json_name, "html": b.html_name, "identity_class": b.identity_class, + "canonical_identity": b.canonical_identity, "display_primary": b.display_primary, + "display_order": b.display_order, "status": b.status, + } + if b.markdown_name is not None: + result["md"] = b.markdown_name + if b.manifest_name is not None: + result["manifest"] = b.manifest_name + return result + + +def test_clean_schema_v1_canonical_and_variant_roundtrip(): + current = candidate(report(generated_at="2026-06-21T00:00:00Z")) + old = candidate(report()) + index = VNextIdentityIndex((binding(old), binding(current, V3_VARIANT, order=1))) + payload = serialize_vnext_index(index) + assert parse_vnext_index(payload) == index + assert serialize_vnext_index(parse_vnext_index(payload)) == payload + assert json.dumps(payload, sort_keys=True, separators=(",", ":")) == json.dumps( + serialize_vnext_index(parse_vnext_index(payload)), sort_keys=True, separators=(",", ":") + ) + + +def test_daily_weekly_monthly_same_as_of_have_distinct_clean_targets(): + candidates = [candidate(report(cadence=cadence)) for cadence in ("daily", "weekly", "monthly")] + index = VNextIdentityIndex(tuple(binding(item, order=position) for position, item in enumerate(candidates))) + names = {item.json_name for item in index.bindings} + assert len(names) == 3 + + +@pytest.mark.parametrize("cadence", ["daily", "weekly", "monthly"]) +def test_all_targets_are_cadence_aware_and_same_date_can_coexist(cadence): + c = candidate(report(cadence=cadence)) + payload = wire(entry(c)) + parsed = parse_vnext_index(payload) + assert parsed.bindings[0].html_name.startswith(f"{c.as_of}-{cadence}-") + assert f"-{cadence}.json" in parsed.bindings[0].json_name + + +def test_variant_uses_full_artifact_digest_and_optional_attachments_are_omittable(): + c = candidate(report(generated_at="2026-06-21T00:00:00Z")) + payload = wire(entry(candidate(report()), md=False), entry(c, V3_VARIANT, md=True, manifest=True, order=1)) + assert len(parse_vnext_index(payload).bindings) == 2 + bad = entry(c, V3_VARIANT, md=True) + bad["md"] = bad["md"].replace(c.artifact_integrity_digest, "0" * 64) + with pytest.raises(VNextIdentityError, match="target_digest_mismatch"): + parse_vnext_index(wire(entry(candidate(report())), bad)) + + +def test_explicit_null_optional_attachment_is_rejected(): + c = candidate(report()) + bad = entry(c) + bad["md"] = None + with pytest.raises(VNextIdentityError, match="invalid_target_name"): + parse_vnext_index(wire(bad)) + + +@pytest.mark.parametrize("field,value", [ + ("namespace", "legacy"), ("schema_version", 2), ("binding_namespace", "legacy"), + ("status", "VERIFIED"), ("report_schema_version", 7), ("semantic_fingerprint_version", "future"), + ("artifact_integrity_version", "future"), ("canonical_identity", False), +]) +def test_unknown_or_forged_contract_fields_fail_closed(field, value): + c = candidate(report()) + payload = wire(entry(c)) + if field in {"namespace", "schema_version"}: + payload[field] = value + else: + payload["reports"][0][field] = value + with pytest.raises(VNextIdentityError): + parse_vnext_index(payload) + + +def test_legacy_filename_and_extra_wire_keys_never_enter_vnext(): + c = candidate(report()) + bad = entry(c) + bad["json"] = f"advisory_report_{c.as_of}.json" + with pytest.raises(VNextIdentityError): + parse_vnext_index(wire(bad)) + bad = entry(c) + bad["debug"] = "legacy" + with pytest.raises(VNextIdentityError): + parse_vnext_index(wire(bad)) + + +def test_exactly_one_canonical_and_display_policy_per_period(): + c = candidate(report()) + variant = candidate(report(generated_at="2026-06-21T00:00:00Z")) + with pytest.raises(VNextIdentityError, match="canonical_missing"): + VNextIdentityIndex((binding(variant, V3_VARIANT),)) + with pytest.raises(VNextIdentityError, match="canonical_conflict"): + VNextIdentityIndex((binding(c), binding(variant, V3_CANONICAL))) + duplicate_primary = binding(variant, V3_VARIANT, primary=True) + with pytest.raises(VNextIdentityError, match="display_primary_conflict"): + VNextIdentityIndex((binding(c, primary=True), duplicate_primary)) + duplicate_order = binding(variant, V3_VARIANT, order=0) + with pytest.raises(VNextIdentityError, match="display_order_conflict"): + VNextIdentityIndex((binding(c), duplicate_order)) + + +def test_display_order_safe_integer_boundaries_and_bool_rejection(): + c = candidate(report()) + assert binding(c, 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(c, order=value) + + +def test_mapping_snapshot_and_serializer_are_deterministic(): + c = candidate(report()) + payload = wire(entry(c)) + ordered = OrderedDict((key, payload[key]) for key in reversed(list(payload))) + assert parse_vnext_index(MappingProxyType(ordered)) == parse_vnext_index(payload) + assert serialize_vnext_index(parse_vnext_index(payload)) == serialize_vnext_index(parse_vnext_index(ordered)) + + +def test_allocation_modes_bootstrap_rerun_exact_and_historical(): + old = candidate(report()) + current = candidate(report(generated_at="2026-06-21T00:00:00Z")) + empty = VNextIdentityIndex(()) + current_context = AllocationContext(AllocationMode.CURRENT_MANDATORY, RequestedArtifacts(), DisplayPlacement(True, 0), old.period_key) + first = allocate_identity(empty, old, current_context) + assert first.binding.identity_class == V3_CANONICAL and not first.reused_existing + rerun = allocate_identity(VNextIdentityIndex((first.binding,)), old, current_context) + assert rerun.reused_existing and rerun.binding == first.binding + changed = allocate_identity(VNextIdentityIndex((first.binding,)), current, current_context) + assert changed.binding.identity_class == V3_VARIANT + with pytest.raises(VNextPublicationError, match="canonical_bootstrap_required"): + allocate_identity(empty, old, AllocationContext(AllocationMode.HISTORICAL_RECOVERY, RequestedArtifacts(), DisplayPlacement(False, 0))) + with pytest.raises(VNextPublicationError, match="identity_reuse_not_found"): + allocate_identity(empty, old, AllocationContext(AllocationMode.EXACT_ARTIFACT_REUSE, RequestedArtifacts(), DisplayPlacement(False, 0))) + + +def test_exact_miss_validates_display_and_policy_before_miss(): + c = candidate(report()) + with pytest.raises(VNextPublicationError, match="display_invalid"): + DisplayPlacement(False, -1) + with pytest.raises(VNextPublicationError, match="identity_reuse_not_found"): + allocate_identity(VNextIdentityIndex(()), c, AllocationContext(AllocationMode.EXACT_ARTIFACT_REUSE, RequestedArtifacts(), DisplayPlacement(False, 0))) + + +def test_publication_plan_uses_binding_targets_not_source_basename(): + c = candidate(report()) + b = binding(c, V3_CANONICAL, primary=True) + plan = build_publication_plan((PublicationEntry(c, b, PublicationRole.MANDATORY_CURRENT, True, 0),)) + assert plan.entries[0].binding.json_name != c.source_identity + assert plan.entries[0].role is PublicationRole.MANDATORY_CURRENT + + +def test_current_variant_is_a_valid_mandatory_publication_entry(): + old = candidate(report()) + current = candidate(report(generated_at="2026-06-21T00:00:00Z")) + old_binding = binding(old) + current_binding = binding(current, V3_VARIANT, order=1) + plan = build_publication_plan(( + PublicationEntry(old, old_binding, PublicationRole.RECOVERED_HISTORY, False, 1), + PublicationEntry(current, current_binding, PublicationRole.MANDATORY_CURRENT, True, 0), + )) + assert plan.entries[0].role is PublicationRole.MANDATORY_CURRENT + assert ".variant-" in plan.entries[0].binding.json_name + + +def test_publication_plan_rejects_collision_and_legacy_shape(): + c = candidate(report()) + b = binding(c) + with pytest.raises(VNextPublicationError, match="publication_target_collision"): + build_publication_plan(( + PublicationEntry(c, b, PublicationRole.MANDATORY_CURRENT, True, 0), + PublicationEntry(c, b, PublicationRole.RECOVERED_HISTORY, False, 1), + )) + with pytest.raises(VNextPublicationError, match="publication_entry_invalid"): + PublicationEntry(c, object(), PublicationRole.MANDATORY_CURRENT, True, 0) From 6db15fab2f28fce8870cdd2d4ee66aa0c346e515 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 04:42:44 +0800 Subject: [PATCH 2/2] fix: close vnext schema v1 contract gaps Co-Authored-By: Codex --- .../vnext_identity_v1.py | 28 +++++++++++++++++-- .../vnext_publication_plan.py | 8 +----- tests/test_vnext_schema_v1.py | 19 ++++++++++++- 3 files changed, 45 insertions(+), 10 deletions(-) diff --git a/src/quant_advisor_research/vnext_identity_v1.py b/src/quant_advisor_research/vnext_identity_v1.py index f49bed0..b214330 100644 --- a/src/quant_advisor_research/vnext_identity_v1.py +++ b/src/quant_advisor_research/vnext_identity_v1.py @@ -195,8 +195,16 @@ def _validate_index(bindings: tuple[VNextIdentityBinding, ...]) -> None: canonical_by_period: set[str] = set() display_by_period: dict[str, tuple[bool, set[int]]] = {} artifact_digests: dict[str, tuple[str, str, str]] = {} + artifact_identities: set[tuple[str, str, str, str, str, str, str]] = set() for binding in bindings: _validate_binding(binding) + artifact_identity = ( + binding.period_key, binding.as_of, binding.cadence, binding.report_schema_version, + binding.contract_version, binding.semantic_digest, binding.artifact_integrity_digest, + ) + if artifact_identity in artifact_identities: + raise _error("identity_duplicate") + artifact_identities.add(artifact_identity) identity = (binding.period_key, binding.json_name, binding.html_name, binding.markdown_name, binding.manifest_name) if binding.canonical_identity: if binding.period_key in canonical_by_period: @@ -272,7 +280,7 @@ def parse_vnext_index(payload: Mapping[str, Any]) -> VNextIdentityIndex: if type(snapshot["reports"]) is not list: raise _error("invalid_wire") try: - bindings = tuple(_entry_from_wire(item) for item in snapshot["reports"]) + bindings = tuple(sorted((_entry_from_wire(item) for item in snapshot["reports"]), key=_binding_sort_key)) return VNextIdentityIndex(bindings) except VNextIdentityError: raise @@ -280,6 +288,21 @@ def parse_vnext_index(payload: Mapping[str, Any]) -> VNextIdentityIndex: raise _error("invalid_wire") 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.as_of, + binding.cadence, + binding.json_name, + binding.html_name, + binding.markdown_name or "", + binding.manifest_name or "", + ) + + def _binding_wire(binding: VNextIdentityBinding) -> dict[str, object]: _validate_binding(binding) result = { @@ -302,8 +325,9 @@ def serialize_vnext_index(index: VNextIdentityIndex) -> dict[str, object]: if not isinstance(index, VNextIdentityIndex): raise _error("identity_index_invalid") _validate_index(index.bindings) + bindings = tuple(sorted(index.bindings, key=_binding_sort_key)) payload = {"schema_version": VNEXT_SCHEMA_VERSION, "namespace": VNEXT_WIRE_NAMESPACE, - "reports": [_binding_wire(binding) for binding in index.bindings]} + "reports": [_binding_wire(binding) for binding in bindings]} parse_vnext_index(payload) return payload diff --git a/src/quant_advisor_research/vnext_publication_plan.py b/src/quant_advisor_research/vnext_publication_plan.py index 12f5b41..9d46be0 100644 --- a/src/quant_advisor_research/vnext_publication_plan.py +++ b/src/quant_advisor_research/vnext_publication_plan.py @@ -254,13 +254,7 @@ def allocate_identity(index: VNextIdentityIndex, candidate: SelectedCandidate, c if context.mode is AllocationMode.HISTORICAL_RECOVERY and not canonical: raise _error("canonical_bootstrap_required") identity_class = V3_VARIANT if canonical else V3_CANONICAL - next_order = max((item.display_order for item in index.bindings if item.period_key == candidate.period_key), default=-1) + 1 - binding = _new_binding( - candidate, - identity_class, - context.requested_artifacts, - DisplayPlacement(False, next_order), - ) + binding = _new_binding(candidate, identity_class, context.requested_artifacts, context.display) _simulate(index, candidate, binding) return AllocationResult(binding, False) diff --git a/tests/test_vnext_schema_v1.py b/tests/test_vnext_schema_v1.py index 8698776..a26ff70 100644 --- a/tests/test_vnext_schema_v1.py +++ b/tests/test_vnext_schema_v1.py @@ -100,6 +100,12 @@ def test_clean_schema_v1_canonical_and_variant_roundtrip(): assert json.dumps(payload, sort_keys=True, separators=(",", ":")) == json.dumps( serialize_vnext_index(parse_vnext_index(payload)), sort_keys=True, separators=(",", ":") ) + reversed_payload = wire( + entry(current, V3_VARIANT, order=1), + entry(old, order=0), + ) + assert parse_vnext_index(payload) == parse_vnext_index(reversed_payload) + assert serialize_vnext_index(parse_vnext_index(payload)) == serialize_vnext_index(parse_vnext_index(reversed_payload)) def test_daily_weekly_monthly_same_as_of_have_distinct_clean_targets(): @@ -179,6 +185,14 @@ def test_exactly_one_canonical_and_display_policy_per_period(): VNextIdentityIndex((binding(c), duplicate_order)) +def test_same_exact_artifact_cannot_have_two_public_identities(): + c = candidate(report()) + canonical = binding(c, md=False) + different_policy = binding(c, md=True) + with pytest.raises(VNextIdentityError, match="identity_duplicate"): + VNextIdentityIndex((canonical, different_policy)) + + def test_display_order_safe_integer_boundaries_and_bool_rejection(): c = candidate(report()) assert binding(c, order=MAX_SAFE_JSON_INTEGER).display_order == MAX_SAFE_JSON_INTEGER @@ -204,7 +218,10 @@ def test_allocation_modes_bootstrap_rerun_exact_and_historical(): assert first.binding.identity_class == V3_CANONICAL and not first.reused_existing rerun = allocate_identity(VNextIdentityIndex((first.binding,)), old, current_context) assert rerun.reused_existing and rerun.binding == first.binding - changed = allocate_identity(VNextIdentityIndex((first.binding,)), current, current_context) + changed = allocate_identity( + VNextIdentityIndex((first.binding,)), current, + AllocationContext(AllocationMode.CURRENT_MANDATORY, RequestedArtifacts(), DisplayPlacement(False, 1), old.period_key), + ) assert changed.binding.identity_class == V3_VARIANT with pytest.raises(VNextPublicationError, match="canonical_bootstrap_required"): allocate_identity(empty, old, AllocationContext(AllocationMode.HISTORICAL_RECOVERY, RequestedArtifacts(), DisplayPlacement(False, 0)))