fix: keep legacy identities out of v3 allocation

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

Author: Pigbibi

src/quant_advisor_research/allocation_context.py

@@ -18,6 +18,7 @@
 from .identity_lifecycle import FINGERPRINT_VERSION, IdentityMetadataError
 from .identity_v3 import (
     PENDING_ARTIFACT_VALIDATION,
+    LEGACY_V2,
     V3_CANONICAL,
     V3_VARIANT,
     V3IdentityBinding,
@@ -329,9 +330,16 @@ def allocate_v3_identity(
         raise _error("allocation_context_mismatch")
     verified_inventory = _revalidate_inventory(inventory)
     bindings = verified_inventory.index.bindings
-    exact_matches = [binding for binding in bindings if _exact_key(binding) == _exact_key_from_metadata(metadata)]
-    same_artifact = [
+    v3_bindings = [
         binding for binding in bindings
+        if binding.identity_class in {V3_CANONICAL, V3_VARIANT}
+    ]
+    exact_matches = [
+        binding for binding in v3_bindings
+        if _exact_key(binding) == _exact_key_from_metadata(metadata)
+    ]
+    same_artifact = [
+        binding for binding in v3_bindings
         if binding.period_key == period_key
         and binding.artifact_integrity_version == ARTIFACT_INTEGRITY_VERSION
         and binding.artifact_integrity_digest == artifact_digest
@@ -343,10 +351,10 @@ def allocate_v3_identity(
         _validate_reuse_policy(binding, requested, display_placement)
         return V3AllocationPlan(binding, context.mode, True)
     if context.mode is AllocationMode.EXACT_ARTIFACT_REUSE:
-        raise _error("identity_reuse_mismatch" if same_artifact else "identity_reuse_not_found")
+        raise _error("exact_artifact_not_found")
     if same_artifact:
         raise _error("identity_reuse_mismatch")
-    period_bindings = [binding for binding in bindings if binding.period_key == period_key]
+    period_bindings = [binding for binding in v3_bindings if binding.period_key == period_key]
     canonical_exists = any(binding.canonical_identity for binding in period_bindings)
     if context.mode is AllocationMode.HISTORICAL_RECOVERY and not canonical_exists:
         raise _error("canonical_bootstrap_required")
@@ -371,9 +379,13 @@ def _exact_key_from_metadata(metadata: tuple[dict[str, object], str, str, str, s
 
 def _simulate(index: V3IdentityIndex, candidate: V3IdentityBinding) -> None:
     try:
+        # LEGACY_V2 entries remain dual-read migration evidence. The candidate
+        # must satisfy the complete v3 ledger invariants without rewriting or
+        # treating legacy canonical metadata as a v3 canonical owner.
+        v3_bindings = tuple(binding for binding in index.bindings if binding.identity_class != LEGACY_V2)
         parse_v3_index({
             "schema_version": 3,
-            "reports": [_binding_payload(binding) for binding in (*index.bindings, candidate)],
+            "reports": [_binding_payload(binding) for binding in (*v3_bindings, candidate)],
         })
     except IdentityMetadataError as exc:
         raise _error(exc.code) from None

tests/test_allocation_context.py

@@ -2,6 +2,7 @@
 
 import hashlib
 from collections import OrderedDict
+from dataclasses import replace
 from pathlib import Path
 from types import MappingProxyType
 
@@ -61,13 +62,15 @@ def binding_for(
     display_order: int = 0,
     include_md: bool = False,
     include_manifest: bool = False,
+    identity_class: str | None = None,
 ):
     period_key, semantic, artifact, contract = metadata(report)
     artifact_digest = artifact_digest or artifact
     semantic_digest = semantic_digest or semantic
     as_of = report["as_of"]
     cadence = report["cadence"]
-    suffix = "" if canonical else f".variant-{artifact_digest}"
+    identity_class = identity_class or ("V3_CANONICAL" if canonical else "V3_VARIANT")
+    suffix = "" if canonical else f".variant-{semantic_digest if identity_class == 'LEGACY_V2' else artifact_digest}"
     entry = {
         "period_key": period_key,
         "as_of": as_of,
@@ -80,7 +83,7 @@ def binding_for(
         "artifact_integrity_digest": artifact_digest,
         "json": f"advisory_report_{as_of}{suffix}.json",
         "html": f"{as_of}-{cadence}-model-recommendations{suffix}.html",
-        "identity_class": "V3_CANONICAL" if canonical else "V3_VARIANT",
+        "identity_class": identity_class,
         "canonical_identity": canonical,
         "display_primary": display_primary,
         "display_order": display_order,
@@ -199,7 +202,7 @@ def test_exact_artifact_reuse_is_immutable_and_requires_full_match() -> None:
 
     changed = dict(report)
     changed["generated_at"] = "2026-07-15T00:00:00Z"
-    with pytest.raises(IdentityMetadataError, match="identity_reuse_not_found|identity_reuse_mismatch"):
+    with pytest.raises(IdentityMetadataError, match="exact_artifact_not_found"):
         allocate_v3_identity(
             changed,
             inventory=inventory,
@@ -276,6 +279,83 @@ def test_inventory_accepts_declared_v3_schema_version() -> None:
     assert inventory.index.schema_version == 3
 
 
+def test_legacy_exact_match_is_not_reused_or_migrated() -> None:
+    report = build_report()
+    legacy = binding_for(report, identity_class="LEGACY_V2")
+
+    with pytest.raises(IdentityMetadataError, match="exact_artifact_not_found"):
+        allocate_v3_identity(
+            report,
+            inventory=inventory_for((legacy, report)),
+            context=AllocationContext.exact_artifact_reuse(),
+            requested_artifacts=REQUESTED,
+            display_placement=DISPLAY,
+        )
+
+
+def test_legacy_only_current_bootstraps_v3_canonical() -> None:
+    report = build_report()
+    legacy = binding_for(report, identity_class="LEGACY_V2")
+    period_key = metadata(report)[0]
+
+    result = allocate_v3_identity(
+        report,
+        inventory=inventory_for((legacy, report)),
+        context=AllocationContext.current_mandatory(period_key),
+        requested_artifacts=REQUESTED,
+        display_placement=DISPLAY,
+    )
+
+    assert result.reused_existing is False
+    assert result.binding.identity_class == "V3_CANONICAL"
+    assert result.binding.canonical_identity is True
+
+
+def test_legacy_only_historical_recovery_requires_v3_canonical() -> None:
+    report = build_report()
+    legacy = binding_for(report, identity_class="LEGACY_V2")
+
+    with pytest.raises(IdentityMetadataError, match="canonical_bootstrap_required"):
+        allocate_v3_identity(
+            report,
+            inventory=inventory_for((legacy, report)),
+            context=AllocationContext.historical_recovery(),
+            requested_artifacts=REQUESTED,
+            display_placement=DISPLAY,
+        )
+
+
+def test_mixed_legacy_and_v3_uses_v3_canonical_for_variant_allocation() -> None:
+    legacy_report = build_report()
+    current_report = dict(legacy_report)
+    current_report["generated_at"] = "2026-07-15T00:00:00Z"
+    candidate_report = dict(current_report)
+    candidate_report["generated_at"] = "2026-07-16T00:00:00Z"
+    legacy_canonical = binding_for(legacy_report, identity_class="LEGACY_V2")
+    legacy = replace(
+        legacy_canonical,
+        canonical_identity=False,
+        json_name=f"advisory_report_{legacy_report['as_of']}.variant-{metadata(legacy_report)[1]}.json",
+        html_name=(
+            f"{legacy_report['as_of']}-{legacy_report['cadence']}-model-recommendations"
+            f".variant-{metadata(legacy_report)[1]}.html"
+        ),
+    )
+    canonical = binding_for(current_report)
+    period_key = metadata(candidate_report)[0]
+
+    result = allocate_v3_identity(
+        candidate_report,
+        inventory=inventory_for((legacy, legacy_report), (canonical, current_report)),
+        context=AllocationContext.current_mandatory(period_key),
+        requested_artifacts=REQUESTED,
+        display_placement=DISPLAY,
+    )
+
+    assert result.binding.identity_class == "V3_VARIANT"
+    assert result.reused_existing is False
+
+
 def test_semantic_same_artifact_different_is_not_reused() -> None:
     report = build_report()
     changed = dict(report)
@@ -284,7 +364,7 @@ def test_semantic_same_artifact_different_is_not_reused() -> None:
     period_key, semantic, artifact, _contract = metadata(changed)
 
     assert semantic == metadata(report)[1]
-    with pytest.raises(IdentityMetadataError, match="identity_reuse_not_found|identity_reuse_mismatch"):
+    with pytest.raises(IdentityMetadataError, match="exact_artifact_not_found"):
         allocate_v3_identity(
             changed,
             inventory=inventory,