FIX: fold non-ASCII case in the seed value search

pyrit/memory/analytics_sql.py

@@ -49,6 +49,24 @@ class UnicodeLower(FunctionElement[str]):
     inherit_cache = True
 
 
+def _sqlite_fold_case(*, value: str, fallback_function: str) -> str:
+    """
+    Keep ASCII text in SQLite as plain ``lower`` and use a registered fold otherwise.
+
+    Byte/character lengths differ for Unicode and embedded NULs. Non-text
+    values keep the Python conversion rather than SQLite's stringification.
+    Both Unicode fold elements share this expression so the BLOB-length guard
+    and the ASCII fast path cannot drift apart.
+
+    Returns:
+        str: The compiled fold expression.
+    """
+    return (
+        f"CASE WHEN typeof({value}) = 'text' AND length(CAST({value} AS BLOB)) = length({value}) "
+        f"THEN lower({value}) ELSE {fallback_function}({value}) END"
+    )
+
+
 @compiles(UnicodeLower, "sqlite")
 def _sqlite_unicode_lower(element: UnicodeLower, compiler: Any, **kwargs: Any) -> str:
     """
@@ -58,12 +76,47 @@ def _sqlite_unicode_lower(element: UnicodeLower, compiler: Any, **kwargs: Any) -
         str: Exact Unicode-aware lowercasing without a Python/GIL callback for ASCII text.
     """
     value = _arguments(element, compiler, **kwargs)[0]
-    # Byte/character lengths differ for Unicode and embedded NULs. Non-text
-    # values keep the existing Python conversion rather than SQLite's stringification.
-    return (
-        f"CASE WHEN typeof({value}) = 'text' AND length(CAST({value} AS BLOB)) = length({value}) "
-        f"THEN lower({value}) ELSE {UnicodeLower.SQLITE_FUNCTION_NAME}({value}) END"
-    )
+    return _sqlite_fold_case(value=value, fallback_function=UnicodeLower.SQLITE_FUNCTION_NAME)
+
+
+class UnicodeCaseFold(FunctionElement[str]):
+    """
+    Fold text per character with Python casefold semantics for exact matching.
+
+    Unlike ``UnicodeLower``, every character maps on its own, so verbatim
+    text always matches regardless of context sensitive forms such as Greek
+    final sigma. Full casefold semantics, including ``ss`` matching the sharp
+    s, hold on SQLite through the registered function; SQL Server compiles to
+    plain ``LOWER``, where such expansions follow the column collation.
+    """
+
+    SQLITE_FUNCTION_NAME: ClassVar[str] = "pyrit_analytics_casefold"
+    type = UnicodeText()
+    inherit_cache = True
+
+
+@compiles(UnicodeCaseFold, "sqlite")
+def _sqlite_unicode_casefold(element: UnicodeCaseFold, compiler: Any, **kwargs: Any) -> str:
+    """
+    Reuse the shared SQLite fold expression with the registered casefold.
+
+    Returns:
+        str: The compiled per character fold expression.
+    """
+    value = _arguments(element, compiler, **kwargs)[0]
+    return _sqlite_fold_case(value=value, fallback_function=UnicodeCaseFold.SQLITE_FUNCTION_NAME)
+
+
+@compiles(UnicodeCaseFold, "mssql")
+def _mssql_unicode_casefold(element: UnicodeCaseFold, compiler: Any, **kwargs: Any) -> str:
+    """
+    Compile the casefold for SQL Server, whose LOWER stays per character.
+
+    Returns:
+        str: The compiled per character fold expression.
+    """
+    value = _arguments(element, compiler, **kwargs)[0]
+    return f"LOWER({value})"
 
 
 @compiles(UnicodeLower, "mssql")

pyrit/memory/attack_analytics_query.py

@@ -55,6 +55,7 @@
     JsonIsArray,
     JsonObjectAggregate,
     JsonScalar,
+    UnicodeCaseFold,
     UnicodeLower,
     UnicodeStrip,
 )
@@ -689,7 +690,7 @@ def _search_condition(*, label: ColumnElement[Any], search: str) -> ColumnElemen
             ColumnElement[bool]: A case-insensitive substring predicate.
         """
         escaped = search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_").replace("[", "\\[")
-        return UnicodeLower(label).like(UnicodeLower(literal(f"%{escaped}%", UnicodeText())), escape="\\")
+        return UnicodeCaseFold(label).like(UnicodeCaseFold(literal(f"%{escaped}%", UnicodeText())), escape="\\")
 
     def results(self, *, limit: int, after: DecodedKeysetCursor | None = None) -> Select[*tuple[Any, ...]]:
         """

pyrit/memory/memory_interface.py

@@ -27,6 +27,7 @@
     Select,
     String,
     Unicode,
+    UnicodeText,
     and_,
     case,
     column,
@@ -60,7 +61,15 @@
 from pyrit.common import print_deprecation_message
 from pyrit.common.async_compatibility import legacy_sync_override, run_legacy_sync_async
 from pyrit.common.pagination import DecodedKeysetCursor
-from pyrit.memory.analytics_sql import JsonArrayItems, JsonArrayJoin, JsonArrayValue, JsonContainer, JsonScalar, UuidHex
+from pyrit.memory.analytics_sql import (
+    JsonArrayItems,
+    JsonArrayJoin,
+    JsonArrayValue,
+    JsonContainer,
+    JsonScalar,
+    UnicodeCaseFold,
+    UuidHex,
+)
 
 if TYPE_CHECKING:
     from pyrit.memory.memory_embedding import MemoryEmbedding
@@ -5480,12 +5489,17 @@ def _seed_example_filters(
             member_conditions.append(SeedEntry.seed_type.in_(list(seed_types)))
         if value_search:
             # A simulated-conversation value is JSON, so its keys would match common words such as "prompt".
+            # casefold maps each character independently, so lowering the term separately from the
+            # stored text cannot lose context sensitive forms such as Greek final sigma, and verbatim
+            # text always still matches.
             pattern = "%" + re.sub(r"([\\%_\[])", r"\\\1", value_search) + "%"
             member_conditions.append(
                 and_(
                     SeedEntry.data_type == "text",
                     SeedEntry.seed_type != "simulated_conversation",
-                    SeedEntry.value.ilike(pattern, escape="\\"),
+                    UnicodeCaseFold(SeedEntry.value).like(
+                        UnicodeCaseFold(literal(pattern, UnicodeText())), escape="\\"
+                    ),
                 )
             )
 

pyrit/memory/sqlite_memory.py

@@ -27,7 +27,7 @@
 
 from pyrit.common import Singleton
 from pyrit.common.path import DB_DATA_PATH
-from pyrit.memory.analytics_sql import UnicodeLower
+from pyrit.memory.analytics_sql import UnicodeCaseFold, UnicodeLower
 from pyrit.memory.memory_interface import MemoryInterface
 from pyrit.memory.memory_models import (
     AttackResultEntry,
@@ -280,15 +280,28 @@ def _unicode_lower(value: str | int | float | bytes | None) -> str | None:
         """
         return str(value).lower() if value is not None else None
 
+    @staticmethod
+    def _unicode_casefold(value: str | int | float | bytes | None) -> str | None:
+        """
+        Casefold SQLite text per character, retaining NULL semantics.
+
+        Returns:
+            str | None: The folded value, or SQL NULL.
+        """
+        return str(value).casefold() if value is not None else None
+
     @staticmethod
     def _register_analytics_lower(*, engine: Engine) -> None:
-        """Install the Unicode lowercase function for analytics and JSON filters on every pooled SQLite connection."""
+        """Install the Unicode fold functions for analytics and JSON filters on every pooled SQLite connection."""
 
         @event.listens_for(engine, "connect")
         def register(dbapi_connection: Any, connection_record: Any) -> None:
             dbapi_connection.create_function(
                 UnicodeLower.SQLITE_FUNCTION_NAME, 1, SQLiteMemory._unicode_lower, deterministic=True
             )
+            dbapi_connection.create_function(
+                UnicodeCaseFold.SQLITE_FUNCTION_NAME, 1, SQLiteMemory._unicode_casefold, deterministic=True
+            )
 
     async def get_session_async(self) -> AsyncSession:
         """

tests/unit/memory/memory_interface/test_interface_seed_examples.py

@@ -160,6 +160,48 @@ async def test_get_seed_examples_harm_filter_folds_non_ascii_case(sqlite_instanc
     assert await _ids(sqlite_instance, harm_categories=["Überwachung"]) == [surveillance]
 
 
+async def test_get_seed_examples_value_search_folds_non_ascii_case(sqlite_instance: MemoryInterface):
+    cafe, wildcard = uuid4(), uuid4()
+    await _add(
+        sqlite_instance,
+        SeedPrompt(value="CAFÉ au lait tactics", dataset_name=DATASET, prompt_group_id=cafe),
+        SeedPrompt(value="50%_off coupon", dataset_name=DATASET, prompt_group_id=wildcard),
+    )
+
+    assert await _ids(sqlite_instance, value_search="CAFÉ") == [cafe]
+    assert await _ids(sqlite_instance, value_search="café") == [cafe]
+    assert await _ids(sqlite_instance, value_search="Au Lait") == [cafe]
+    assert await _ids(sqlite_instance, value_search="50%_off") == [wildcard]
+
+
+async def test_get_seed_examples_value_search_matches_greek_final_sigma(sqlite_instance: MemoryInterface):
+    """Verbatim Greek text must match regardless of context sensitive final sigma."""
+    astro = uuid4()
+    await _add(
+        sqlite_instance,
+        SeedPrompt(value="ΑΣΤΡΟ navigation charts", dataset_name=DATASET, prompt_group_id=astro),
+    )
+
+    # 'ΑΣ' lower()s to 'ας' while the stored text lower()s to 'αστρο', so naive
+    # lowering of the two sides separately loses the match; casefold keeps it.
+    assert await _ids(sqlite_instance, value_search="ΑΣ") == [astro]
+    assert await _ids(sqlite_instance, value_search="ασ") == [astro]
+
+
+async def test_get_seed_examples_value_search_casefold_expansions_hold_on_sqlite(
+    sqlite_instance: MemoryInterface,
+):
+    """SQLite's registered casefold maps the sharp s to ss, so both spellings match."""
+    strasse = uuid4()
+    await _add(
+        sqlite_instance,
+        SeedPrompt(value="Straße crossing signs", dataset_name=DATASET, prompt_group_id=strasse),
+    )
+
+    assert await _ids(sqlite_instance, value_search="STRASSE") == [strasse]
+    assert await _ids(sqlite_instance, value_search="straße") == [strasse]
+
+
 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/test_analytics_migration.py

@@ -15,14 +15,15 @@
 from alembic.config import Config
 from alembic.operations import Operations
 from alembic.script import ScriptDirectory
-from sqlalchemy import String, create_engine, inspect, text
+from sqlalchemy import String, UnicodeText, create_engine, inspect, literal, text
 from sqlalchemy.dialects import mssql, sqlite
 from sqlalchemy.engine import Connection
 from sqlalchemy.schema import CreateIndex, CreateTable
 
 from pyrit.memory import AttackResultEntry
 from pyrit.memory.alembic.versions.e5f7a9c1b3d2_add_identifiers_tables import IdentifierGraphInserter
 from pyrit.memory.analytics_identity_v1 import ObjectiveTargetAnalyticsIdentityV1
+from pyrit.memory.analytics_sql import UnicodeCaseFold
 from pyrit.models import AtomicAttackIdentifier, AttackIdentifier, TargetIdentifier, compute_eval_hash
 
 if TYPE_CHECKING:
@@ -324,6 +325,43 @@ def test_analytics_migration_preserves_outcome_at_the_index_key_limit(tmp_path:
         engine.dispose()
 
 
+def test_unicode_casefold_compiles_per_dialect() -> None:
+    """SQLite keeps the registered casefold fallback and SQL Server uses LOWER exactly."""
+    expression = UnicodeCaseFold(literal("CAFÉ", UnicodeText()))
+    sqlite_sql = str(expression.compile(dialect=sqlite.dialect()))
+    mssql_sql = str(expression.compile(dialect=mssql.dialect()))
+    assert "pyrit_analytics_casefold" in sqlite_sql
+    assert "lower(" in sqlite_sql  # the ASCII fast path stays plain lower
+    # Pinned exactly: a wrapper around the fallback would change Azure folding.
+    assert mssql_sql == "LOWER(:param_1)"
+
+
+def test_seed_value_search_folds_both_sides_in_the_real_statement() -> None:
+    """The seed search statement itself folds column and pattern in SQL."""
+    from sqlalchemy import true
+
+    from pyrit.memory.sqlite_memory import SQLiteMemory
+
+    conditions = SQLiteMemory._seed_example_filters(
+        SQLiteMemory.__new__(SQLiteMemory),
+        scope=true(),
+        data_types=None,
+        harm_categories=None,
+        seed_types=None,
+        value_search="CAFÉ",
+    )
+    assert conditions, "the value search must produce a condition"
+    for dialect, fold_marker in (
+        (sqlite.dialect(), "pyrit_analytics_casefold"),
+        (mssql.dialect(), "LOWER"),
+    ):
+        for condition in conditions:
+            compiled = str(condition.compile(dialect=dialect))
+            # Column and pattern both fold in SQL, so neither backend can mix
+            # foldings by lowering one side in Python.
+            assert compiled.count(fold_marker) == 2, compiled
+
+
 @pytest.mark.parametrize("dialect", [sqlite.dialect(), mssql.dialect()])
 def test_generated_lookup_and_index_ddl_are_dialect_specific(dialect: Dialect) -> None:
     sql = str(CreateTable(AttackResultEntry.__table__).compile(dialect=dialect))

tests/unit/memory/test_attack_analytics.py

@@ -236,6 +236,32 @@ async def test_non_ascii_case_variants_share_group_filter_and_facet_keys(sqlite_
         assert (await connection.execute(select(UnicodeLower(literal("\u00c9COLE"))))).scalar_one() == "\u00e9cole"
 
 
+@pytest.mark.parametrize("search", ["\u0391\u03a3", "\u039f\u03a3"])
+@pytest.mark.parametrize(
+    "dimension",
+    [
+        AttackAnalyticsDimension(name="label", label_key="case"),
+        AttackAnalyticsDimension(name="targeted_harm_category"),
+        AttackAnalyticsDimension(name="attack_type"),
+    ],
+    ids=["label", "category", "attack_type"],
+)
+async def test_facet_search_matches_greek_sigma_verbatim(
+    *, sqlite_instance: SQLiteMemory, dimension: AttackAnalyticsDimension, search: str
+) -> None:
+    """Facet search folds per character, so final sigma never hides an exact match."""
+    astros = "\u0391\u03a3\u03a4\u03a1\u039f\u03a3"
+    await sqlite_instance.add_attack_results_to_memory_async(
+        attack_results=[
+            make_result(labels={"case": astros}, categories=[astros], attack_class_name=astros),
+            make_result(index=2, labels={"case": "other"}, categories=["other"], attack_class_name="Other"),
+        ]
+    )
+    query = AttackAnalyticsFacetQuery(dimension=dimension, search=search)
+    facet = await AttackAnalyticsReader(memory=sqlite_instance).facets_async(query=query, control=control())
+    assert [item.label.casefold() for item in facet.items] == [astros.casefold()]
+
+
 @pytest.mark.parametrize(
     ("value", "fallback"),
     [