FIX: match non-ASCII values in SQLite seed filters

Author: bIackr0se

pyrit/memory/memory_interface.py

@@ -4000,7 +4000,8 @@ def _add_list_conditions(
                 )
             )
         else:
-            conditions.extend(field.contains(value) for value in values)
+            # The column holds JSON text with non-ASCII characters escaped, so search for the value as JSON spells it.
+            conditions.extend(field.contains(json.dumps(value)[1:-1]) for value in values)
 
     async def _serialize_seed_value_async(self, prompt: Seed) -> str:
         """

pyrit/memory/sqlite_memory.py

@@ -273,7 +273,7 @@ def _unicode_lower(value: str | int | float | bytes | None) -> str | None:
 
     @staticmethod
     def _register_analytics_lower(*, engine: Engine) -> None:
-        """Install the analytics-only Unicode function on every pooled SQLite connection."""
+        """Install the Unicode lowercase function for analytics and JSON filters on every pooled SQLite connection."""
 
         @event.listens_for(engine, "connect")
         def register(dbapi_connection: Any, connection_record: Any) -> None:
@@ -447,7 +447,7 @@ def _get_condition_json_property_match(
         if case_sensitive:
             extracted_value, target = raw, value
         else:
-            extracted_value, target = func.lower(raw), value.lower()
+            extracted_value, target = UnicodeLower(raw), value.lower()
 
         if partial_match:
             escaped = target.replace("%", "\\%").replace("_", "\\_")
@@ -494,7 +494,9 @@ def _get_condition_json_array_match(
         column_name = json_column.key
         pp_param = f"property_path_{uid}"
         sp_param = f"array_element_path_{uid}"
-        value_expression = f"LOWER(json_extract(value, :{sp_param}))" if array_element_path else "LOWER(value)"
+        element = f"json_extract(value, :{sp_param})" if array_element_path else "value"
+        # SQLite's LOWER folds ASCII only; match the Unicode folding applied to the values below.
+        value_expression = f"{UnicodeLower.SQLITE_FUNCTION_NAME}({element})"
 
         conditions = []
         bindparams_dict: dict[str, str] = {pp_param: property_path}

tests/unit/memory/memory_interface/test_interface_remove_seeds.py

@@ -275,6 +275,21 @@ async def test_remove_seeds_by_list_filters_is_case_insensitive(sqlite_instance:
     assert await sqlite_instance.remove_seeds_from_memory_async(harm_categories=["violence"]) == 1
 
 
+async def test_remove_seeds_by_list_filters_matches_non_ascii_values(sqlite_instance: MemoryInterface):
+    seed_prompts = [
+        SeedPrompt(value="p1", harm_categories=["ÉTHIQUE"], data_type="text"),
+        SeedPrompt(value="p2", harm_categories=["Überwachung"], data_type="text"),
+        SeedPrompt(value="p3", harm_categories=["fraud"], data_type="text"),
+    ]
+    await sqlite_instance.add_seeds_to_memory_async(seeds=seed_prompts, added_by="test")
+
+    assert await sqlite_instance.remove_seeds_from_memory_async(harm_categories=["ÉTHIQUE"]) == 1
+    assert await sqlite_instance.remove_seeds_from_memory_async(harm_categories=["überwachung"]) == 1
+
+    remaining = await sqlite_instance.get_seeds_async()
+    assert [seed.value for seed in remaining] == ["p3"]
+
+
 async def test_remove_seeds_by_groups_authors_parameters_match_whole_elements(sqlite_instance: MemoryInterface):
     seed_prompts = [
         SeedPrompt(value="p1", groups=["team"], authors=["Ann"], parameters=["goal"], data_type="text"),

tests/unit/memory/memory_interface/test_interface_seed_examples.py

@@ -148,6 +148,19 @@ async def test_get_seed_examples_filters_match_across_members(sqlite_instance: M
     assert await _ids(sqlite_instance, seed_types=["objective"], data_types=["image_path"]) == []
 
 
+async def test_get_seed_examples_harm_filter_folds_non_ascii_case(sqlite_instance: MemoryInterface):
+    ethics, surveillance = uuid4(), uuid4()
+    await _add(
+        sqlite_instance,
+        SeedPrompt(value="a", dataset_name=DATASET, prompt_group_id=ethics, harm_categories=["ÉTHIQUE"]),
+        SeedPrompt(value="b", dataset_name=DATASET, prompt_group_id=surveillance, harm_categories=["überwachung"]),
+    )
+
+    assert await _ids(sqlite_instance, harm_categories=["ÉTHIQUE"]) == [ethics]
+    assert await _ids(sqlite_instance, harm_categories=["éthique"]) == [ethics]
+    assert await _ids(sqlite_instance, harm_categories=["Überwachung"]) == [surveillance]
+
+
 async def test_get_seed_examples_keeps_named_and_unnamed_scopes_apart(sqlite_instance: MemoryInterface):
     group = uuid4()
     named = SeedPrompt(value="named", dataset_name=DATASET, prompt_group_id=group)

tests/unit/memory/memory_interface/test_interface_seed_prompts.py

@@ -522,6 +522,31 @@ async def test_get_seeds_with_substring_filters_harm_categories(sqlite_instance:
     assert result[1].authors == ["author2"]
 
 
+@pytest.mark.parametrize(
+    "filters, expected",
+    [
+        ({"authors": ["Frédéric Dubut"]}, ["p1"]),
+        ({"authors": ["Frédéric"]}, ["p1"]),
+        ({"harm_categories": ["ÉTHIQUE"]}, ["p1"]),
+        ({"harm_categories": ["Überwachung"]}, ["p2"]),
+        ({"authors": ['"AJ"']}, ["p3"]),
+        ({"authors": ["C:\\red"]}, ["p3"]),
+    ],
+)
+async def test_get_seeds_with_list_filters_matches_non_ascii_values(
+    sqlite_instance: MemoryInterface, filters: dict[str, list[str]], expected: list[str]
+):
+    seed_prompts = [
+        SeedPrompt(value="p1", harm_categories=["ÉTHIQUE"], authors=["Frédéric Dubut"], data_type="text"),
+        SeedPrompt(value="p2", harm_categories=["Überwachung"], authors=["Frederic Smith"], data_type="text"),
+        SeedPrompt(value="p3", authors=['Ann "AJ" Jones', "C:\\red-team"], data_type="text"),
+    ]
+    await sqlite_instance.add_seeds_to_memory_async(seeds=seed_prompts, added_by="test")
+
+    result = await sqlite_instance.get_seeds_async(**filters)
+    assert [seed.value for seed in result] == expected
+
+
 async def test_get_seeds_with_substring_filters_groups(sqlite_instance: MemoryInterface):
     seed_prompts = [
         SeedPrompt(value="prompt1", groups=["group1"], data_type="text"),

tests/unit/memory/test_sqlite_memory.py

@@ -1248,3 +1248,37 @@ async def test_get_message_pieces_filters_on_integer_prompt_metadata_async(sqlit
 
     assert len(retrieved) == 1
     assert retrieved[0].original_value == "sent"
+
+
+@pytest.mark.parametrize(
+    "kwargs, expected",
+    [
+        ({"property_path": "$.label", "value": "éthique"}, 1),
+        ({"property_path": "$.label", "value": "ÉTHIQUE"}, 1),
+        ({"property_path": "$.label", "value": "éth", "partial_match": True}, 1),
+        ({"property_path": "$.label", "value": "éthique", "case_sensitive": True}, 0),
+        ({"property_path": "$.label", "value": "ethique"}, 0),
+        ({"property_path": "$.items", "array_element_path": "$.name", "array_to_match": ["ÜBERWACHUNG", "fraud"]}, 1),
+        ({"property_path": "$.items", "array_element_path": "$.name", "array_to_match": ["überwachung", "missing"]}, 0),
+    ],
+)
+async def test_json_match_conditions_fold_non_ascii_case_async(
+    sqlite_instance: SQLiteMemory, kwargs: dict[str, object], expected: int
+) -> None:
+    piece = MessagePiece(
+        conversation_id=str(uuid.uuid4()),
+        role="assistant",
+        original_value="stored",
+        prompt_metadata={"label": "ÉTHIQUE", "items": [{"name": "Überwachung"}, {"name": "Fraud"}]},
+    )
+    await sqlite_instance.add_message_pieces_to_memory_async(message_pieces=[piece])
+
+    match = (
+        sqlite_instance._get_condition_json_array_match
+        if "array_to_match" in kwargs
+        else sqlite_instance._get_condition_json_property_match
+    )
+    condition = match(json_column=PromptMemoryEntry.prompt_metadata, **kwargs)
+    async with await sqlite_instance.get_session_async() as session:
+        rows = (await session.execute(select(PromptMemoryEntry.id).where(condition))).all()
+    assert len(rows) == expected