fix: scope identity digest conflicts by period

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

Author: Pigbibi

src/quant_advisor_research/publication_identity.py

@@ -221,18 +221,20 @@ def parse_reports_index(payload: object) -> ReportsIndex:
         bindings = tuple(_validate_v2_entry(entry) for entry in reports)
     else:
         raise _error("unsupported_index_version")
-    identity_map: dict[tuple[str, str, str | None, str | None], str | None] = {}
-    digest_map: dict[str, tuple[str, str, str | None, str | None]] = {}
+    identity_map: dict[tuple[str, tuple[str, str, str | None, str | None]], str | None] = {}
+    digest_map: dict[tuple[str, str, str], tuple[str, str, str | None, str | None]] = {}
     for binding in bindings:
         identity = (binding.json_name, binding.html_name, binding.markdown_name, binding.manifest_name)
-        if identity in identity_map and identity_map[identity] != binding.fingerprint_digest:
+        identity_key = (binding.period_key, identity)
+        if identity_key in identity_map and identity_map[identity_key] != binding.fingerprint_digest:
             raise _error("identity_content_conflict")
-        identity_map[identity] = binding.fingerprint_digest
+        identity_map[identity_key] = binding.fingerprint_digest
         if binding.fingerprint_digest is not None:
-            previous = digest_map.get(binding.fingerprint_digest)
+            digest_key = (binding.period_key, binding.fingerprint_version or "", binding.fingerprint_digest)
+            previous = digest_map.get(digest_key)
             if previous is not None and previous != identity:
                 raise _error("identity_digest_conflict")
-            digest_map[binding.fingerprint_digest] = identity
+            digest_map[digest_key] = identity
     return ReportsIndex(schema_version=schema_version, bindings=bindings)
 
 
@@ -280,6 +282,15 @@ def serialize_reports_index_v2(index: ReportsIndex) -> str:
         ),
     )
     for binding in ordered:
+        if (
+            binding.schema_version not in {"5", "6"}
+            or binding.fingerprint_version != FINGERPRINT_VERSION
+            or binding.fingerprint_digest is None
+            or type(binding.display_order) is not int
+            or type(binding.canonical_identity) is not bool
+            or type(binding.display_primary) is not bool
+        ):
+            raise _error("v2_serialization_invalid_binding")
         item: dict[str, Any] = {
             "period_key": binding.period_key,
             "as_of": binding.as_of,
@@ -297,5 +308,9 @@ def serialize_reports_index_v2(index: ReportsIndex) -> str:
             item["md"] = binding.markdown_name
         if binding.manifest_name is not None:
             item["manifest"] = binding.manifest_name
+        try:
+            _validate_v2_entry(item)
+        except IdentityMetadataError as exc:
+            raise _error("v2_serialization_invalid_binding") from exc
         reports.append(item)
     return json.dumps({"schema_version": 2, "reports": reports}, ensure_ascii=False, indent=2) + "\n"

tests/test_publication_identity.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import hashlib
+import json
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -192,6 +194,68 @@ def test_v2_duplicate_identity_or_digest_conflicts_fail_closed() -> None:
         parse_reports_index({"schema_version": 2, "reports": [first, same_digest]})
 
 
+def test_same_digest_is_allowed_across_different_periods() -> None:
+    first_report = build_report(as_of="2026-06-20")
+    second_report = build_report(as_of="2026-06-27")
+    digest = digest_for(first_report)
+
+    index = parse_reports_index(
+        {
+            "schema_version": 2,
+            "reports": [v2_entry(first_report, digest=digest), v2_entry(second_report, digest=digest)],
+        }
+    )
+
+    assert {binding.period_key for binding in index.bindings} == {
+        "weekly:2026-06-15:2026-06-21",
+        "weekly:2026-06-22:2026-06-28",
+    }
+
+
+def test_v1_verified_binding_cannot_be_serialized_as_incomplete_v2() -> None:
+    report = build_report()
+    binding = parse_reports_index({"schema_version": 1, "reports": [v1_entry()]}).bindings[0]
+    verified_v1 = verify_identity_binding(binding, report)
+
+    with pytest.raises(IdentityMetadataError, match="v2_serialization_invalid_binding"):
+        serialize_reports_index_v2(ReportsIndex(2, (verified_v1,)))
+
+
+def test_complete_v2_binding_serializes_and_parses_round_trip() -> None:
+    report = build_report()
+    parsed = parse_reports_index({"schema_version": 2, "reports": [v2_entry(report)]})
+    verified = verify_identity_binding(parsed.bindings[0], report)
+
+    round_tripped = parse_reports_index(
+        json.loads(serialize_reports_index_v2(ReportsIndex(2, (verified,))))
+    ).bindings[0]
+
+    assert round_tripped.period_key == verified.period_key
+    assert round_tripped.fingerprint_digest == verified.fingerprint_digest
+    assert round_tripped.json_name == verified.json_name
+    assert round_tripped.verification_status == "PENDING_REPORT_VALIDATION"
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda binding: replace(binding, display_order=None),
+        lambda binding: replace(binding, display_order=-1),
+        lambda binding: replace(binding, display_order=True),
+        lambda binding: replace(binding, fingerprint_version=None),
+        lambda binding: replace(binding, fingerprint_digest=None),
+        lambda binding: replace(binding, schema_version=None),
+    ],
+)
+def test_serializer_revalidates_complete_v2_invariants(mutation) -> None:
+    report = build_report()
+    parsed = parse_reports_index({"schema_version": 2, "reports": [v2_entry(report)]})
+    verified = verify_identity_binding(parsed.bindings[0], report)
+
+    with pytest.raises(IdentityMetadataError, match="v2_serialization"):
+        serialize_reports_index_v2(ReportsIndex(2, (mutation(verified),)))
+
+
 @pytest.mark.parametrize("key", ["md", "manifest"])
 def test_optional_identity_names_are_validated_when_declared(key: str) -> None:
     report = build_report()