feat: add converged clean-slate vNext identity allocation

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/qar_vnext_identity_contract.md

@@ -0,0 +1,30 @@
+# QAR clean-slate vNext N1 identity contract
+
+The independent wire namespace is exactly
+`qar_vnext.identity_index.v1` with top-level keys `schema_version` and
+`entries`. It accepts only `V3_CANONICAL` and `V3_VARIANT`; legacy identity
+classes, old index versions, migration and fallback are rejected.
+
+Every public target is cadence-aware:
+
+- canonical: `advisory_report_<as_of>-<cadence>.(json|md)` and
+  `<as_of>-<cadence>-model-recommendations.html`;
+- variant: the same names with
+  `.variant-<full artifact_integrity_digest>` before the extension;
+- manifest uses the matching JSON stem and suffix.
+
+Optional `md` and `manifest` are presence-sensitive: omission means the target
+was not requested; a present key must contain a valid basename string. `null`,
+empty strings and other types are invalid and cannot be silently dropped by
+serialization.
+
+`EXACT_ARTIFACT_REUSE`, `CURRENT_MANDATORY` and `HISTORICAL_RECOVERY` validate
+all inputs before lookup. An identical current artifact reuses its immutable
+canonical binding; a changed artifact in an occupied period receives a full
+artifact-digest variant. Current publication display is separate from stored
+identity, while exact reuse requires stored display equality and historical
+reuse does not treat display as identity.
+
+This foundation is pure: it performs no filesystem, network, publisher,
+build, workflow, Pages, migration or production I/O. The old runtime/index
+path is isolated and is not a vNext compatibility mode.

src/quant_advisor_research/publication_plan.py

@@ -27,6 +27,7 @@
 from .period_contract import PeriodContractError, canonical_period_identity
 from .publisher import report_content_fingerprint
 from .time_contract import TimeContractError, contract_version_for_schema
+from .vnext_binding import VNextBindingError, validate_vnext_binding
 
 
 class PublicationPlanError(ValueError):
@@ -173,6 +174,7 @@ class PublicationEntry:
     role: PublicationRole
     display_primary: bool
     display_order: int
+    identity_namespace: str = "legacy"
 
     def __post_init__(self) -> None:
         _validate_entry(self)
@@ -227,17 +229,24 @@ def _binding_payload(binding: V3IdentityBinding) -> dict[str, object]:
     return payload
 
 
-def _validate_binding(binding: object) -> V3IdentityBinding:
+def _validate_binding(binding: object, *, strict_vnext: bool) -> V3IdentityBinding:
     if not isinstance(binding, V3IdentityBinding):
         raise _error("identity_binding_invalid")
     if binding.status != PENDING_ARTIFACT_VALIDATION:
         raise _error("identity_binding_invalid")
-    if binding.identity_class == LEGACY_V2:
+    if strict_vnext and binding.identity_class == LEGACY_V2:
+        raise _error("legacy_binding_not_migrated")
+    if not strict_vnext and binding.identity_class == LEGACY_V2:
         raise _error("legacy_binding_not_migrated")
-    if binding.identity_class not in {V3_CANONICAL, V3_VARIANT}:
-        raise _error("identity_binding_invalid")
     try:
-        validated = _validate_v3_entry(_binding_payload(binding))
+        if strict_vnext:
+            validated = validate_vnext_binding(binding)
+        else:
+            if binding.identity_class not in {V3_CANONICAL, V3_VARIANT}:
+                raise _error("identity_binding_invalid")
+            validated = _validate_v3_entry(_binding_payload(binding))
+    except VNextBindingError as exc:
+        raise _error(exc.code) from None
     except (IdentityMetadataError, AttributeError, KeyError, TypeError, ValueError, OverflowError, UnicodeError):
         raise _error("identity_binding_invalid") from None
     if validated != binding:
@@ -252,9 +261,11 @@ def _validate_entry(entry: PublicationEntry) -> None:
         raise _error("candidate_invalid")
     if type(entry.role) is not PublicationRole:
         raise _error("publication_role_invalid")
+    if type(entry.identity_namespace) is not str or entry.identity_namespace not in {"legacy", "vnext"}:
+        raise _error("identity_namespace_invalid")
     if type(entry.display_primary) is not bool or type(entry.display_order) is not int or entry.display_order < 0:
         raise _error("display_evidence_invalid")
-    binding = _validate_binding(entry.binding)
+    binding = _validate_binding(entry.binding, strict_vnext=entry.identity_namespace == "vnext")
     candidate = entry.candidate
     try:
         rebuilt = SelectedCandidate.from_report(

src/quant_advisor_research/vnext_binding.py

@@ -0,0 +1,157 @@
+"""Strict shared validator for the clean-slate vNext identity namespace."""
+
+from __future__ import annotations
+
+import re
+
+from .artifact_integrity import ARTIFACT_INTEGRITY_VERSION
+from .identity_lifecycle import FINGERPRINT_VERSION
+from .identity_v3 import PENDING_ARTIFACT_VALIDATION, V3_CANONICAL, V3_VARIANT, V3IdentityBinding
+from .period_contract import PeriodContractError, canonical_period_identity
+from .time_contract import TimeContractError, contract_version_for_schema
+
+
+class VNextBindingError(ValueError):
+    """Stable, sanitized clean-slate binding error."""
+
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
+
+
+_DATE = r"(?P<as_of>\d{4}-\d{2}-\d{2})"
+_DIGEST = r"(?P<digest>[0-9a-f]{64})"
+_CADENCE = r"(?P<cadence>daily|weekly|monthly)"
+_JSON = re.compile(rf"^advisory_report_{_DATE}-{_CADENCE}(?:\.variant-{_DIGEST})?\.json$")
+_HTML = re.compile(rf"^{_DATE}-{_CADENCE}-model-recommendations(?:\.variant-{_DIGEST})?\.html$")
+_MD = re.compile(rf"^advisory_report_{_DATE}-{_CADENCE}(?:\.variant-{_DIGEST})?\.md$")
+_MANIFEST = re.compile(
+    rf"^advisory_report_{_DATE}-{_CADENCE}(?:\.variant-{_DIGEST})?\.json\.manifest\.json$"
+)
+
+
+def _error(code: str) -> VNextBindingError:
+    return VNextBindingError(code)
+
+
+def binding_payload(binding: V3IdentityBinding) -> dict[str, object]:
+    payload: dict[str, object] = {
+        "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,
+    }
+    if binding.markdown_name is not None:
+        payload["md"] = binding.markdown_name
+    if binding.manifest_name is not None:
+        payload["manifest"] = binding.manifest_name
+    return payload
+
+
+def _name_digest(name: object, pattern: re.Pattern[str], *, as_of: str, cadence: str) -> str | None:
+    if type(name) is not str or not name or "/" in name or "\\" in name:
+        raise _error("identity_name_invalid")
+    match = pattern.fullmatch(name)
+    if match is None or match.group("as_of") != as_of or match.group("cadence") != cadence:
+        raise _error("identity_name_mismatch")
+    return match.groupdict().get("digest")
+
+
+def validate_vnext_binding(entry: object) -> V3IdentityBinding:
+    """Validate a clean wire entry or binding; no legacy compatibility exists."""
+
+    if isinstance(entry, V3IdentityBinding):
+        if entry.status != PENDING_ARTIFACT_VALIDATION:
+            raise _error("identity_binding_invalid")
+        entry = binding_payload(entry)
+    required = {
+        "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",
+    }
+    optional = {"md", "manifest"}
+    if type(entry) is not dict or not required.issubset(entry) or set(entry) - required - optional:
+        raise _error("identity_binding_invalid")
+    as_of = entry["as_of"]
+    cadence = entry["cadence"]
+    if type(as_of) is not str or type(cadence) is not str:
+        raise _error("identity_binding_invalid")
+    try:
+        period_key = canonical_period_identity(cadence, as_of).key
+    except (PeriodContractError, TypeError, ValueError, OverflowError):
+        raise _error("period_mismatch") from None
+    if type(entry["period_key"]) is not str or entry["period_key"] != period_key:
+        raise _error("period_mismatch")
+
+    schema = entry["report_schema_version"]
+    contract = entry["contract_version"]
+    if type(schema) is not str:
+        raise _error("invalid_schema_version")
+    if type(contract) is not str:
+        raise _error("invalid_contract_version")
+    try:
+        expected_contract = contract_version_for_schema(schema)
+    except (TimeContractError, TypeError, ValueError):
+        raise _error("invalid_schema_version") from None
+    if contract != expected_contract:
+        raise _error("contract_version_mismatch")
+
+    if entry["semantic_fingerprint_version"] != FINGERPRINT_VERSION:
+        raise _error("invalid_fingerprint_version")
+    semantic_digest = entry["semantic_digest"]
+    if type(semantic_digest) is not str or re.fullmatch(r"[0-9a-f]{64}", semantic_digest) is None:
+        raise _error("invalid_semantic_digest")
+    if entry["artifact_integrity_version"] != ARTIFACT_INTEGRITY_VERSION:
+        raise _error("invalid_artifact_integrity_version")
+    artifact_digest = entry["artifact_integrity_digest"]
+    if type(artifact_digest) is not str or re.fullmatch(r"[0-9a-f]{64}", artifact_digest) is None:
+        raise _error("invalid_artifact_integrity_digest")
+
+    identity_class = entry["identity_class"]
+    if type(identity_class) is not str or identity_class not in {V3_CANONICAL, V3_VARIANT}:
+        raise _error("legacy_identity_rejected" if identity_class == "LEGACY_V2" else "invalid_identity_class")
+    canonical = entry["canonical_identity"]
+    primary = entry["display_primary"]
+    order = entry["display_order"]
+    if type(canonical) is not bool or type(primary) is not bool or type(order) is not int or order < 0:
+        raise _error("identity_binding_invalid")
+    if (identity_class == V3_CANONICAL) != canonical:
+        raise _error("identity_metadata_mismatch")
+
+    names = [
+        (_name_digest(entry["json"], _JSON, as_of=as_of, cadence=cadence), entry["json"]),
+        (_name_digest(entry["html"], _HTML, as_of=as_of, cadence=cadence), entry["html"]),
+    ]
+    if "md" in entry and type(entry["md"]) is not str:
+        raise _error("identity_name_invalid")
+    if "manifest" in entry and type(entry["manifest"]) is not str:
+        raise _error("identity_name_invalid")
+    markdown = entry["md"] if "md" in entry else None
+    manifest = entry["manifest"] if "manifest" in entry else None
+    if markdown is not None:
+        names.append((_name_digest(markdown, _MD, as_of=as_of, cadence=cadence), markdown))
+    if manifest is not None:
+        names.append((_name_digest(manifest, _MANIFEST, as_of=as_of, cadence=cadence), manifest))
+    expected_suffix = None if canonical else artifact_digest
+    if any(name_digest != expected_suffix for name_digest, _name in names):
+        raise _error("identity_digest_mismatch")
+    return V3IdentityBinding(
+        period_key, as_of, cadence, schema, contract, FINGERPRINT_VERSION, semantic_digest,
+        ARTIFACT_INTEGRITY_VERSION, artifact_digest, entry["json"], entry["html"], markdown, manifest,
+        identity_class, canonical, primary, order, PENDING_ARTIFACT_VALIDATION,
+    )
+
+
+__all__ = ["VNextBindingError", "binding_payload", "validate_vnext_binding"]