fix: harden identity trust and inventory

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

Author: Pigbibi

src/quant_advisor_research/identity_lifecycle.py

@@ -2,7 +2,8 @@
 
 import hashlib
 import re
-from dataclasses import dataclass, field
+import weakref
+from dataclasses import dataclass
 from typing import Any
 
 from .contracts import AdvisoryValidationError, validate_advisory_report
@@ -32,12 +33,6 @@ def __init__(self, code: str) -> None:
         super().__init__(code)
 
 
-@dataclass(frozen=True, slots=True)
-class _EvidenceSeal:
-    kind: str
-    payload: tuple[object, ...]
-
-
 @dataclass(frozen=True, slots=True)
 class V1ProvisionalBinding:
     period_key: str
@@ -48,7 +43,7 @@ class V1ProvisionalBinding:
     status: str = "PROVISIONAL"
 
 
-@dataclass(frozen=True, slots=True)
+@dataclass(frozen=True, slots=True, weakref_slot=True)
 class VerifiedReportEvidence:
     period_key: str
     as_of: str
@@ -57,7 +52,6 @@ class VerifiedReportEvidence:
     fingerprint_version: str
     fingerprint_digest: str
     status: str = "VERIFIED_REPORT_EVIDENCE"
-    _seal: _EvidenceSeal | None = field(default=None, repr=False, compare=False)
 
 
 @dataclass(frozen=True, slots=True)
@@ -90,11 +84,10 @@ class V2IdentityIndex:
     bindings: tuple[V2IdentityBinding, ...]
 
 
-@dataclass(frozen=True, slots=True)
+@dataclass(frozen=True, slots=True, weakref_slot=True)
 class VerifiedIdentityEvidence:
     binding: V2IdentityBinding
     report_digest: str
-    _seal: _EvidenceSeal | None = field(default=None, repr=False, compare=False)
 
 
 @dataclass(frozen=True, slots=True)
@@ -113,6 +106,31 @@ class AllocatedIdentity:
     allocation_source: str
 
 
+@dataclass(frozen=True, slots=True, weakref_slot=True)
+class CompleteVerifiedIdentityInventory:
+    identities: tuple[VerifiedIdentityEvidence, ...]
+
+
+_REPORT_CAPABILITIES: dict[int, weakref.ReferenceType[VerifiedReportEvidence]] = {}
+_IDENTITY_CAPABILITIES: dict[int, weakref.ReferenceType[VerifiedIdentityEvidence]] = {}
+_INVENTORY_CAPABILITIES: dict[int, weakref.ReferenceType[CompleteVerifiedIdentityInventory]] = {}
+
+
+def _register_capability(registry: dict[int, weakref.ReferenceType[Any]], value: Any) -> None:
+    key = id(value)
+
+    def discard(reference: weakref.ReferenceType[Any], *, key: int = key) -> None:
+        if registry.get(key) is reference:
+            registry.pop(key, None)
+
+    registry[key] = weakref.ref(value, discard)
+
+
+def _has_capability(registry: dict[int, weakref.ReferenceType[Any]], value: Any) -> bool:
+    reference = registry.get(id(value))
+    return reference is not None and reference() is value
+
+
 def _error(code: str) -> IdentityMetadataError:
     return IdentityMetadataError(code)
 
@@ -306,8 +324,10 @@ def make_verified_report_evidence(
         raise _error("invalid_schema_version")
     if type(fingerprint_digest) is not str or re.fullmatch(r"[0-9a-f]{64}", fingerprint_digest) is None:
         raise _error("invalid_fingerprint_digest")
-    payload = (period_key, as_of, cadence, schema_version, FINGERPRINT_VERSION, fingerprint_digest)
-    return VerifiedReportEvidence(*payload, _seal=_EvidenceSeal("report", payload))
+    return VerifiedReportEvidence(
+        period_key, as_of, cadence, schema_version, FINGERPRINT_VERSION, fingerprint_digest,
+        status="UNTRUSTED_REPORT_EVIDENCE_CANDIDATE",
+    )
 
 
 def _v1_binding_payload(binding: V1ProvisionalBinding) -> dict[str, object]:
@@ -368,14 +388,6 @@ def _compute_report_evidence(report: object, provisional: V1ProvisionalBinding |
     return period_key, as_of, cadence, schema_version, digest
 
 
-def _valid_report_seal(evidence: VerifiedReportEvidence) -> bool:
-    payload = (
-        evidence.period_key, evidence.as_of, evidence.cadence, evidence.schema_version,
-        evidence.fingerprint_version, evidence.fingerprint_digest,
-    )
-    return isinstance(evidence._seal, _EvidenceSeal) and evidence._seal == _EvidenceSeal("report", payload)
-
-
 def verify_report_evidence(
     report: object,
     *,
@@ -384,14 +396,15 @@ def verify_report_evidence(
 ) -> VerifiedReportEvidence:
     period_key, as_of, cadence, schema_version, digest = _compute_report_evidence(report, provisional)
     if expected is not None:
-        if not isinstance(expected, VerifiedReportEvidence) or not _valid_report_seal(expected):
+        if not isinstance(expected, VerifiedReportEvidence):
             raise _error("report_evidence_untrusted")
         if (period_key, as_of, cadence, schema_version, digest) != (
             expected.period_key, expected.as_of, expected.cadence, expected.schema_version, expected.fingerprint_digest
         ):
             raise _error("identity_content_conflict")
-    payload = (period_key, as_of, cadence, schema_version, FINGERPRINT_VERSION, digest)
-    return VerifiedReportEvidence(*payload, _seal=_EvidenceSeal("report", payload))
+    evidence = VerifiedReportEvidence(period_key, as_of, cadence, schema_version, FINGERPRINT_VERSION, digest)
+    _register_capability(_REPORT_CAPABILITIES, evidence)
+    return evidence
 
 
 def verify_existing_identity(report: object, binding: V2IdentityBinding) -> VerifiedIdentityEvidence:
@@ -408,16 +421,40 @@ def verify_existing_identity(report: object, binding: V2IdentityBinding) -> Veri
         raise _error("identity_metadata_mismatch")
     if digest != validated.fingerprint_digest:
         raise _error("identity_content_conflict")
-    payload = (validated, digest)
-    return VerifiedIdentityEvidence(validated, digest, _EvidenceSeal("identity", payload))
+    evidence = VerifiedIdentityEvidence(validated, digest)
+    _register_capability(_IDENTITY_CAPABILITIES, evidence)
+    return evidence
 
 
 def _valid_identity_evidence(evidence: VerifiedIdentityEvidence) -> bool:
-    if not isinstance(evidence, VerifiedIdentityEvidence):
-        return False
-    return isinstance(evidence._seal, _EvidenceSeal) and evidence._seal == _EvidenceSeal(
-        "identity", (evidence.binding, evidence.report_digest)
-    )
+    return isinstance(evidence, VerifiedIdentityEvidence) and _has_capability(_IDENTITY_CAPABILITIES, evidence)
+
+
+def make_complete_identity_inventory(
+    index: V2IdentityIndex,
+    identities: tuple[VerifiedIdentityEvidence, ...] | list[VerifiedIdentityEvidence],
+) -> CompleteVerifiedIdentityInventory:
+    if not isinstance(index, V2IdentityIndex) or index.schema_version != 2:
+        raise _error("identity_inventory_invalid")
+    if any(not _valid_identity_evidence(item) for item in identities):
+        raise _error("identity_evidence_untrusted")
+    bindings = tuple(index.bindings)
+    try:
+        _validate_v2_index(bindings)
+    except IdentityMetadataError:
+        raise
+    evidence = tuple(identities)
+    if len(evidence) != len(bindings):
+        raise _error("identity_inventory_incomplete")
+    expected = sorted(bindings, key=lambda item: (item.period_key, item.json_name, item.html_name))
+    actual = sorted((item.binding for item in evidence), key=lambda item: (item.period_key, item.json_name, item.html_name))
+    if actual != expected:
+        raise _error("identity_inventory_incomplete")
+    inventory = CompleteVerifiedIdentityInventory(tuple(sorted(evidence, key=lambda item: (
+        item.binding.period_key, item.binding.json_name, item.binding.html_name
+    ))))
+    _register_capability(_INVENTORY_CAPABILITIES, inventory)
+    return inventory
 
 
 def _new_allocated_identity(evidence: VerifiedReportEvidence, *, canonical: bool, source: str) -> AllocatedIdentity:
@@ -434,16 +471,21 @@ def _new_allocated_identity(evidence: VerifiedReportEvidence, *, canonical: bool
 def allocate_identity(
     evidence: VerifiedReportEvidence,
     *,
-    existing_identities: tuple[VerifiedIdentityEvidence, ...] | list[VerifiedIdentityEvidence] = (),
+    inventory: CompleteVerifiedIdentityInventory | None = None,
+    existing_identities: object = None,
     current_period_key: str | None = None,
 ) -> AllocatedIdentity:
-    if not isinstance(evidence, VerifiedReportEvidence) or not _valid_report_seal(evidence):
+    if existing_identities is not None:
+        raise _error("identity_inventory_required")
+    if not isinstance(evidence, VerifiedReportEvidence) or not _has_capability(_REPORT_CAPABILITIES, evidence):
         raise _error("report_evidence_untrusted")
+    if not isinstance(inventory, CompleteVerifiedIdentityInventory) or not _has_capability(
+        _INVENTORY_CAPABILITIES, inventory
+    ):
+        raise _error("identity_inventory_required")
     if current_period_key is not None and not isinstance(current_period_key, str):
         raise _error("period_mismatch")
-    identities = tuple(existing_identities)
-    if any(not _valid_identity_evidence(item) for item in identities):
-        raise _error("identity_evidence_untrusted")
+    identities = inventory.identities
     bindings = tuple(item.binding for item in identities)
     try:
         _validate_v2_index(bindings)

tests/test_identity_lifecycle_l2.py

@@ -2,6 +2,9 @@
 
 import hashlib
 import json
+import gc
+import pickle
+import weakref
 from dataclasses import replace
 from pathlib import Path
 
@@ -13,7 +16,9 @@
     IdentityMetadataError,
     V1ProvisionalBinding,
     V2IdentityBinding,
+    V2IdentityIndex,
     allocate_identity,
+    make_complete_identity_inventory,
     make_verified_report_evidence,
     parse_v1_index,
     parse_v2_index,
@@ -153,6 +158,49 @@ def test_forged_or_replaced_evidence_cannot_be_allocated() -> None:
         allocate_identity(forged, current_period_key=evidence.period_key)
 
 
+def test_public_candidate_requires_explicit_report_revalidation_for_trust() -> None:
+    report = build_report()
+    candidate = make_verified_report_evidence(
+        as_of=report["as_of"], cadence=report["cadence"], schema_version=report["schema_version"],
+        fingerprint_digest=digest_for(report),
+    )
+    assert candidate.status == "UNTRUSTED_REPORT_EVIDENCE_CANDIDATE"
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, ()), ())
+
+    with pytest.raises(IdentityMetadataError, match="report_evidence_untrusted"):
+        allocate_identity(candidate, inventory=inventory, current_period_key=candidate.period_key)
+
+    trusted = verify_report_evidence(report, expected=candidate)
+    assert allocate_identity(trusted, inventory=inventory, current_period_key=trusted.period_key).canonical_identity
+
+
+def test_equal_copy_and_pickle_roundtrip_do_not_retain_ephemeral_trust() -> None:
+    trusted = verify_report_evidence(build_report())
+    copied = replace(trusted)
+    unpickled = pickle.loads(pickle.dumps(trusted))
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, ()), ())
+
+    for candidate in (copied, unpickled):
+        with pytest.raises(IdentityMetadataError, match="report_evidence_untrusted"):
+            allocate_identity(candidate, inventory=inventory)
+
+
+def test_legacy_subset_argument_fails_closed() -> None:
+    trusted = verify_report_evidence(build_report())
+
+    with pytest.raises(IdentityMetadataError, match="identity_inventory_required"):
+        allocate_identity(trusted, existing_identities=[])
+
+
+def test_report_capability_registry_does_not_keep_evidence_alive() -> None:
+    trusted = verify_report_evidence(build_report())
+    reference = weakref.ref(trusted)
+    del trusted
+    gc.collect()
+
+    assert reference() is None
+
+
 def test_existing_identity_requires_report_bytes_and_digest_match() -> None:
     first = build_report()
     binding = v2_binding(first)
@@ -171,8 +219,9 @@ def test_exact_verified_identity_is_reused_immutably() -> None:
     report = build_report()
     evidence = verify_report_evidence(report)
     existing = verify_existing_identity(report, v2_binding(report))
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, (existing.binding,)), (existing,))
 
-    allocated = allocate_identity(evidence, existing_identities=[existing])
+    allocated = allocate_identity(evidence, inventory=inventory)
 
     assert allocated.json_name == existing.binding.json_name
     assert allocated.html_name == existing.binding.html_name
@@ -181,8 +230,9 @@ def test_exact_verified_identity_is_reused_immutably() -> None:
 
 def test_current_period_without_existing_canonical_allocates_canonical() -> None:
     evidence = verify_report_evidence(build_report())
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, ()), ())
 
-    allocated = allocate_identity(evidence, current_period_key=evidence.period_key)
+    allocated = allocate_identity(evidence, inventory=inventory, current_period_key=evidence.period_key)
 
     assert allocated.canonical_identity is True
     assert ".variant-" not in allocated.json_name
@@ -193,8 +243,9 @@ def test_current_period_with_existing_different_canonical_allocates_variant() ->
     new_report = replace_report_content(old_report)
     old_identity = verify_existing_identity(old_report, v2_binding(old_report))
     new_evidence = verify_report_evidence(new_report)
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, (old_identity.binding,)), (old_identity,))
 
-    allocated = allocate_identity(new_evidence, existing_identities=[old_identity], current_period_key=new_evidence.period_key)
+    allocated = allocate_identity(new_evidence, inventory=inventory, current_period_key=new_evidence.period_key)
 
     assert allocated.canonical_identity is False
     assert f".variant-{new_evidence.fingerprint_digest}" in allocated.json_name
@@ -203,8 +254,9 @@ def test_current_period_with_existing_different_canonical_allocates_variant() ->
 
 def test_historical_new_content_always_allocates_variant() -> None:
     evidence = verify_report_evidence(build_report())
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, ()), ())
 
-    allocated = allocate_identity(evidence)
+    allocated = allocate_identity(evidence, inventory=inventory)
 
     assert allocated.canonical_identity is False
     assert f".variant-{evidence.fingerprint_digest}" in allocated.html_name
@@ -224,8 +276,45 @@ def test_allocation_is_input_order_independent() -> None:
     )
     second_identity = verify_existing_identity(second, second_binding)
     evidence = verify_report_evidence(build_report(as_of="2026-06-27"))
+    inventory = make_complete_identity_inventory(
+        V2IdentityIndex(2, (first_identity.binding, second_identity.binding)),
+        [first_identity, second_identity],
+    )
 
-    one = allocate_identity(evidence, existing_identities=[first_identity, second_identity])
-    two = allocate_identity(evidence, existing_identities=[second_identity, first_identity])
+    one = allocate_identity(evidence, inventory=inventory)
+    two = allocate_identity(evidence, inventory=inventory)
 
     assert one == two
+
+
+def test_complete_inventory_requires_one_to_one_full_index_coverage() -> None:
+    report = build_report()
+    existing = verify_existing_identity(report, v2_binding(report))
+    full_index = V2IdentityIndex(2, (existing.binding,))
+
+    with pytest.raises(IdentityMetadataError, match="identity_inventory_incomplete"):
+        make_complete_identity_inventory(full_index, ())
+    with pytest.raises(IdentityMetadataError, match="identity_inventory_incomplete"):
+        make_complete_identity_inventory(V2IdentityIndex(2, ()), (existing,))
+
+
+def test_complete_inventory_rejects_duplicate_or_filtered_bindings() -> None:
+    report = build_report()
+    existing = verify_existing_identity(report, v2_binding(report))
+    duplicate_index = V2IdentityIndex(2, (existing.binding, existing.binding))
+
+    with pytest.raises(IdentityMetadataError, match="identity_content_conflict"):
+        make_complete_identity_inventory(duplicate_index, (existing, existing))
+
+
+def test_rehydrated_identity_evidence_requires_binding_revalidation() -> None:
+    report = build_report()
+    trusted = verify_existing_identity(report, v2_binding(report))
+    reconstructed = pickle.loads(pickle.dumps(trusted))
+
+    with pytest.raises(IdentityMetadataError, match="identity_evidence_untrusted"):
+        make_complete_identity_inventory(V2IdentityIndex(2, (trusted.binding,)), (reconstructed,))
+
+    revalidated = verify_existing_identity(report, trusted.binding)
+    inventory = make_complete_identity_inventory(V2IdentityIndex(2, (revalidated.binding,)), (revalidated,))
+    assert allocate_identity(verify_report_evidence(report), inventory=inventory).json_name == revalidated.binding.json_name