fix: validate allocation reuse policy

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

Author: Pigbibi

src/quant_advisor_research/allocation_context.py

@@ -22,6 +22,7 @@
     V3_VARIANT,
     V3IdentityBinding,
     V3IdentityIndex,
+    V3_SCHEMA_VERSION,
     parse_v3_index,
 )
 from .period_contract import canonical_period_identity
@@ -127,12 +128,19 @@ def _binding_payload(binding: V3IdentityBinding) -> dict[str, object]:
 
 
 def _validated_index(index: object) -> V3IdentityIndex:
-    if not isinstance(index, V3IdentityIndex) or type(index.schema_version) is not int:
+    if (
+        not isinstance(index, V3IdentityIndex)
+        or type(index.schema_version) is not int
+        or index.schema_version != V3_SCHEMA_VERSION
+    ):
         raise _error("identity_inventory_invalid")
     try:
         if type(index.bindings) is not tuple:
             raise _error("identity_inventory_invalid")
-        payload = {"schema_version": 3, "reports": [_binding_payload(binding) for binding in index.bindings]}
+        payload = {
+            "schema_version": index.schema_version,
+            "reports": [_binding_payload(binding) for binding in index.bindings],
+        }
         validated = parse_v3_index(payload)
         if len(validated.bindings) != len(index.bindings):
             raise _error("identity_inventory_invalid")
@@ -258,6 +266,21 @@ def _validate_display(display: object) -> DisplayPlacement:
     return display
 
 
+def _validate_reuse_policy(
+    binding: V3IdentityBinding,
+    requested: RequestedArtifactSet,
+    display_placement: object,
+) -> None:
+    display = _validate_display(display_placement)
+    if (
+        (binding.markdown_name is not None) != requested.include_markdown
+        or (binding.manifest_name is not None) != requested.include_manifest
+        or binding.display_primary != display.display_primary
+        or binding.display_order != display.display_order
+    ):
+        raise _error("identity_reuse_mismatch")
+
+
 def _exact_key(binding: V3IdentityBinding) -> tuple[str, ...]:
     return (
         binding.period_key, binding.as_of, binding.cadence, binding.report_schema_version,
@@ -317,6 +340,7 @@ def allocate_v3_identity(
         raise _error("identity_integrity_conflict")
     if exact_matches:
         binding = exact_matches[0]
+        _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")

tests/test_allocation_context.py

@@ -189,6 +189,7 @@ def test_exact_artifact_reuse_is_immutable_and_requires_full_match() -> None:
         inventory=inventory,
         context=AllocationContext.exact_artifact_reuse(),
         requested_artifacts=RequestedArtifactSet(False, False),
+        display_placement=DisplayPlacement(True, 2),
     )
 
     assert result.reused_existing is True
@@ -204,9 +205,77 @@ def test_exact_artifact_reuse_is_immutable_and_requires_full_match() -> None:
             inventory=inventory,
             context=AllocationContext.exact_artifact_reuse(),
             requested_artifacts=REQUESTED,
+            display_placement=DISPLAY,
+        )
+
+
+@pytest.mark.parametrize(
+    ("requested", "display"),
+    [
+        (RequestedArtifactSet(True, False), DisplayPlacement(True, 2)),
+        (RequestedArtifactSet(False, True), DisplayPlacement(True, 2)),
+        (RequestedArtifactSet(False, False), DisplayPlacement(False, 2)),
+        (RequestedArtifactSet(False, False), DisplayPlacement(True, 3)),
+    ],
+)
+def test_exact_reuse_rejects_publication_policy_mismatch(
+    requested: RequestedArtifactSet, display: DisplayPlacement
+) -> None:
+    report = build_report()
+    binding = binding_for(report, display_primary=True, display_order=2)
+
+    with pytest.raises(IdentityMetadataError, match="identity_reuse_mismatch"):
+        allocate_v3_identity(
+            report,
+            inventory=inventory_for((binding, report)),
+            context=AllocationContext.exact_artifact_reuse(),
+            requested_artifacts=requested,
+            display_placement=display,
+        )
+
+
+@pytest.mark.parametrize(
+    "display",
+    [None, DisplayPlacement(True, True), DisplayPlacement(True, -1)],
+)
+def test_exact_reuse_requires_valid_matching_display_placement(display: object) -> None:
+    report = build_report()
+    binding = binding_for(report, display_primary=True, display_order=2)
+
+    with pytest.raises(IdentityMetadataError, match="display_placement"):
+        allocate_v3_identity(
+            report,
+            inventory=inventory_for((binding, report)),
+            context=AllocationContext.exact_artifact_reuse(),
+            requested_artifacts=REQUESTED,
+            display_placement=display,  # type: ignore[arg-type]
+        )
+
+
+@pytest.mark.parametrize("schema_version", [2, 4, True, "3"])
+def test_inventory_rejects_declared_schema_version_other_than_v3(schema_version: object) -> None:
+    report = build_report()
+    binding = binding_for(report)
+
+    with pytest.raises(IdentityMetadataError, match="identity_inventory_invalid"):
+        make_complete_source_inventory(
+            V3IdentityIndex(schema_version, (binding,)),  # type: ignore[arg-type]
+            {binding.json_name: report},
         )
 
 
+def test_inventory_accepts_declared_v3_schema_version() -> None:
+    report = build_report()
+    binding = binding_for(report)
+
+    inventory = make_complete_source_inventory(
+        V3IdentityIndex(3, (binding,)),
+        {binding.json_name: report},
+    )
+
+    assert inventory.index.schema_version == 3
+
+
 def test_semantic_same_artifact_different_is_not_reused() -> None:
     report = build_report()
     changed = dict(report)
@@ -279,6 +348,7 @@ def test_display_and_identity_ownership_are_independent() -> None:
         inventory=inventory,
         context=AllocationContext.exact_artifact_reuse(),
         requested_artifacts=REQUESTED,
+        display_placement=DisplayPlacement(False, 0),
     )
 
     assert result.binding.canonical_identity is True