fix: resolve simprint type mismatch in iscc_id queries, cap simprints per type

iscc_search/indexes/common.py

@@ -10,11 +10,16 @@
 
 import re
 import json
+import random
 import iscc_core as ic
-from iscc_search.schema import IsccEntry
+from loguru import logger
+from iscc_search.schema import IsccEntry, IsccQuery
 from iscc_search.models import IsccUnit, IsccCode
 
 
+MAX_SIMPRINTS_PER_TYPE = 20
+
+
 # Validation patterns
 INDEX_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9]*$")
 
@@ -212,6 +217,57 @@ def validate_iscc_id(iscc_id, expected_realm=None):
         )
 
 
+def query_from_asset(asset):
+    # type: (IsccEntry) -> IsccQuery
+    """
+    Build an IsccQuery from a stored IsccEntry.
+
+    Converts IsccSimprint objects to bare base64 strings expected by IsccQuery.
+
+    :param asset: Stored asset entry
+    :return: IsccQuery with iscc_code, units, and simprints from the asset
+    """
+    simprints = None
+    if asset.simprints:
+        simprints = {sp_type: [sp.simprint for sp in sp_list] for sp_type, sp_list in asset.simprints.items()}
+    return IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=simprints)
+
+
+def cap_simprints(query):
+    # type: (IsccQuery) -> IsccQuery
+    """
+    Cap simprints to MAX_SIMPRINTS_PER_TYPE per type via random sampling.
+
+    When a query contains more than MAX_SIMPRINTS_PER_TYPE simprints for any type,
+    a random subset is selected to keep search load bounded. This provides a
+    representative sample of the content rather than biasing toward any position.
+
+    :param query: Query that may contain simprints
+    :return: Query with simprints capped (unchanged if already within limit)
+    """
+    if not query.simprints:
+        return query
+
+    capped = {}
+    changed = False
+    for sp_type, sp_list in query.simprints.items():
+        if len(sp_list) > MAX_SIMPRINTS_PER_TYPE:
+            capped[sp_type] = random.sample(sp_list, MAX_SIMPRINTS_PER_TYPE)
+            logger.info(
+                "Capped {} simprints from {} to {} (random sample)",
+                sp_type,
+                len(sp_list),
+                MAX_SIMPRINTS_PER_TYPE,
+            )
+            changed = True
+        else:
+            capped[sp_type] = sp_list
+
+    if changed:
+        return query.model_copy(update={"simprints": capped})
+    return query
+
+
 def normalize_query(query):
     # type: (IsccQuery) -> IsccQuery
     """
@@ -229,16 +285,22 @@ def normalize_query(query):
     4. If query has only simprints → return as is (simprints-only query)
     5. If query has neither units, iscc_code, nor simprints → raise error
 
+    Simprints are capped to MAX_SIMPRINTS_PER_TYPE per type via random sampling
+    to keep search load bounded.
+
     Note: Not all unit combinations form valid ISCC-CODEs. Units-only queries
     that don't form valid codes (e.g., missing DATA/INSTANCE) will work with
     LMDB backend but return no matches with memory backend.
 
     :param query: Query object (may have iscc_code, units, simprints, or combination)
-    :return: Query with units and/or iscc_code populated (simprints passed through)
+    :return: Query with units and/or iscc_code populated, simprints capped
     :raises ValueError: If query has neither iscc_code, units, nor simprints
     """
     import iscc_core as ic
 
+    # Cap simprints before further processing
+    query = cap_simprints(query)
+
     # Case 1: Has both - return as is
     if query.units and query.iscc_code:
         return query

iscc_search/indexes/lmdb/index.py

@@ -208,12 +208,8 @@ def search_assets(self, query, limit=100):
         query_iscc_id = None  # Track original query iscc_id for self-exclusion
         if query.iscc_id:
             query_iscc_id = query.iscc_id
-            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
             asset = self.get_asset(query.iscc_id)
-            # Create new query with extracted iscc_code, units and simprints
-            from iscc_search.schema import IsccQuery
-
-            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+            query = common.query_from_asset(asset)
 
         # Normalize query to ensure it has units (derive from iscc_code if needed)
         query = common.normalize_query(query)

iscc_search/indexes/memory/index.py

@@ -194,12 +194,8 @@ def search_assets(self, index_name, query, limit=100):
         query_iscc_id = None  # Track original query iscc_id for self-exclusion
         if query.iscc_id:
             query_iscc_id = query.iscc_id
-            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
             asset = self.get_asset(index_name, query.iscc_id)
-            # Create new query with extracted iscc_code, units and simprints
-            from iscc_search.schema import IsccQuery
-
-            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+            query = common.query_from_asset(asset)
 
         # Normalize query to ensure it has units (derive from iscc_code if needed)
         # This ensures consistent behavior across backends

iscc_search/indexes/usearch/index.py

@@ -481,12 +481,8 @@ def search_assets(self, query, limit=100, exact=False):
         query_iscc_id = None  # Track original query iscc_id for self-exclusion
         if query.iscc_id:
             query_iscc_id = query.iscc_id
-            # Look up asset by iscc_id (raises FileNotFoundError if not found -> HTTP 404)
             asset = self.get_asset(query.iscc_id)
-            # Create new query with extracted iscc_code, units and simprints
-            from iscc_search.schema import IsccQuery
-
-            query = IsccQuery(iscc_code=asset.iscc_code, units=asset.units, simprints=asset.simprints)
+            query = common.query_from_asset(asset)
 
         # Normalize query
         query = common.normalize_query(query)

tests/test_indexes_common.py

@@ -6,7 +6,7 @@
 
 import pytest
 import iscc_core as ic
-from iscc_search.schema import IsccEntry, IsccQuery
+from iscc_search.schema import IsccEntry, IsccQuery, IsccSimprint
 from iscc_search.indexes import common
 
 
@@ -288,6 +288,67 @@ def test_roundtrip_iscc_id_body_reconstruction(sample_iscc_ids):
     assert reconstructed == original_iscc_id
 
 
+def test_query_from_asset_without_simprints(sample_iscc_codes):
+    """Test query_from_asset converts asset without simprints."""
+    asset = IsccEntry(iscc_code=sample_iscc_codes[0])
+    query = common.query_from_asset(asset)
+
+    assert query.iscc_code == sample_iscc_codes[0]
+    assert query.simprints is None
+
+
+def test_query_from_asset_with_simprints(sample_iscc_codes):
+    """Test query_from_asset converts IsccSimprint objects to base64 strings."""
+    sp = IsccSimprint(simprint="AXvu3tp2kF8mN9qL4rT1sZ", offset=0, size=350)
+    asset = IsccEntry(
+        iscc_code=sample_iscc_codes[0],
+        simprints={"CONTENT_TEXT_V0": [sp]},
+    )
+    query = common.query_from_asset(asset)
+
+    assert query.iscc_code == sample_iscc_codes[0]
+    assert query.simprints is not None
+    assert query.simprints["CONTENT_TEXT_V0"][0].root == "AXvu3tp2kF8mN9qL4rT1sZ"
+
+
+def test_cap_simprints_no_simprints():
+    """Test cap_simprints passes through query without simprints."""
+    query = IsccQuery(iscc_code="ISCC:KACT4EBWK27737D2AYCJRAL5Z36G76RFRMO4554RQ55OSRMQSJVQ")
+    result = common.cap_simprints(query)
+    assert result.simprints is None
+
+
+def test_cap_simprints_within_limit():
+    """Test cap_simprints passes through simprints within limit."""
+    sps = [f"AXvu3tp2kF8m{i:04d}" for i in range(10)]
+    query = IsccQuery(simprints={"CONTENT_TEXT_V0": sps})
+    result = common.cap_simprints(query)
+    assert len(result.simprints["CONTENT_TEXT_V0"]) == 10
+
+
+def test_cap_simprints_over_limit():
+    """Test cap_simprints randomly samples when over MAX_SIMPRINTS_PER_TYPE."""
+    sps = [f"AXvu3tp2kF8m{i:04d}" for i in range(50)]
+    query = IsccQuery(simprints={"CONTENT_TEXT_V0": sps})
+    result = common.cap_simprints(query)
+    capped = result.simprints["CONTENT_TEXT_V0"]
+    assert len(capped) == common.MAX_SIMPRINTS_PER_TYPE
+    # All sampled simprints must come from the original set
+    original_roots = {sp.root if hasattr(sp, "root") else sp for sp in sps}
+    for sp in capped:
+        assert (sp.root if hasattr(sp, "root") else sp) in original_roots
+
+
+def test_cap_simprints_mixed_types():
+    """Test cap_simprints caps per type independently."""
+    small = [f"AXvu3tp2kF8m{i:04d}" for i in range(5)]
+    large = [f"BYwu4uq3lG9n{i:04d}" for i in range(30)]
+    query = IsccQuery(simprints={"CONTENT_TEXT_V0": small, "CONTENT_IMAGE_V0": large})
+    result = common.cap_simprints(query)
+    assert len(result.simprints["CONTENT_TEXT_V0"]) == 5
+    assert len(result.simprints["CONTENT_IMAGE_V0"]) == common.MAX_SIMPRINTS_PER_TYPE
+
+
 def test_normalize_query_with_iscc_code_only(sample_iscc_codes):
     """Test normalize_query derives units from iscc_code."""
     query = IsccQuery(iscc_code=sample_iscc_codes[0])