Merge: merge develop branch to middle branch

backend/alembic/versions/00091_add_conversation_search_indexes.py

@@ -1,15 +1,19 @@
 """add indexes to speed up conversation search/filter
 
-Targets the slow conversation search queries (topic/summary ILIKE, correlated
-EXISTS on conversation_analysis, ORDER BY created_at DESC pagination).
+Targets the slow conversation search queries (correlated EXISTS on
+conversation_analysis, ORDER BY created_at DESC pagination).
 
 Adds:
 - b-tree index on conversation_analysis.conversation_id (FK had no index -> the
   correlated EXISTS subquery was doing a seq scan per conversation row)
-- pg_trgm GIN indexes on conversations.topic and conversation_analysis.summary so
-  leading-wildcard ILIKE '%x%' can use an index instead of a seq scan
 - partial b-tree on conversations.created_at DESC (live rows) for ORDER BY + LIMIT
 
+Note: topic/summary ILIKE '%x%' searches are intentionally NOT indexed here.
+A leading-wildcard ILIKE can only be index-backed via a pg_trgm GIN index, and
+creating that extension requires CREATE privilege on the database which is not
+available in all (e.g. managed prod) environments. Those searches continue to
+run as sequential scans.
+
 All indexes are built with CREATE INDEX CONCURRENTLY so no write locks are taken
 on prod tables. CONCURRENTLY cannot run inside a transaction, so we use an
 autocommit block (Alembic wraps each migration in a transaction by default).
@@ -32,10 +36,6 @@
 
 
 def upgrade() -> None:
-    # pg_trgm can be created inside the normal transaction; it enables the
-    # gin_trgm_ops operator class used by the topic/summary indexes below.
-    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
-
     # CREATE INDEX CONCURRENTLY must run outside a transaction.
     with op.get_context().autocommit_block():
         # (1) Critical: index the FK column the correlated EXISTS subquery uses.
@@ -47,23 +47,7 @@ def upgrade() -> None:
             """
         )
 
-        # (2) Trigram indexes so ILIKE '%term%' is index-backed.
-        op.execute(
-            """
-            CREATE INDEX CONCURRENTLY IF NOT EXISTS
-                ix_conversations_topic_trgm
-            ON conversations USING gin (topic gin_trgm_ops)
-            """
-        )
-        op.execute(
-            """
-            CREATE INDEX CONCURRENTLY IF NOT EXISTS
-                ix_conversation_analysis_summary_trgm
-            ON conversation_analysis USING gin (summary gin_trgm_ops)
-            """
-        )
-
-        # (3) Support ORDER BY created_at DESC + pagination for live rows.
+        # (2) Support ORDER BY created_at DESC + pagination for live rows.
         op.execute(
             """
             CREATE INDEX CONCURRENTLY IF NOT EXISTS
@@ -80,13 +64,6 @@ def downgrade() -> None:
         op.execute(
             "DROP INDEX CONCURRENTLY IF EXISTS ix_conversations_created_at_desc"
         )
-        op.execute(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_conversation_analysis_summary_trgm"
-        )
-        op.execute(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_conversations_topic_trgm"
-        )
         op.execute(
             "DROP INDEX CONCURRENTLY IF EXISTS ix_conversation_analysis_conversation_id"
         )
-    # Leave the pg_trgm extension in place; other objects may depend on it.

backend/alembic/versions/00095_add_conversation_to_test_cases.py

@@ -0,0 +1,55 @@
+"""add_conversation_to_test_cases
+
+Links each test case to the conversation it was imported from and its position
+within that conversation, so evaluation runs can replay each conversation as an
+isolated memory thread.
+
+Both columns are nullable: legacy and manually created cases have no source
+conversation and are treated as independent single-turn conversations.
+
+Revision ID: 4e1c7a9d2b05
+Revises: 322883fb3a4d
+Create Date: 2026-07-20 10:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "4e1c7a9d2b05"
+down_revision: Union[str, None] = "322883fb3a4d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "test_cases",
+        sa.Column("source_conversation_id", postgresql.UUID(as_uuid=True), nullable=True),
+    )
+    op.add_column(
+        "test_cases",
+        sa.Column("turn_index", sa.Integer(), nullable=True),
+    )
+    op.create_index(
+        "ix_test_cases_source_conversation_id",
+        "test_cases",
+        ["source_conversation_id"],
+    )
+    # Execution groups by conversation and orders by turn within it.
+    op.create_index(
+        "ix_test_cases_conversation_turn",
+        "test_cases",
+        ["suite_id", "source_conversation_id", "turn_index"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_test_cases_conversation_turn", table_name="test_cases")
+    op.drop_index("ix_test_cases_source_conversation_id", table_name="test_cases")
+    op.drop_column("test_cases", "turn_index")
+    op.drop_column("test_cases", "source_conversation_id")

backend/alembic/versions/00096_backfill_test_case_conversations.py

@@ -0,0 +1,169 @@
+"""backfill_test_case_conversations
+
+Recovers the source conversation of previously imported test cases.
+
+A suite's imported cases are treated as one cluster and linked only when exactly
+one live conversation yields precisely the same ordered question/answer exchanges,
+rebuilt with the same speaker roles and message order the importer used. Anything
+ambiguous, partial or unmatched stays null and is logged for manual review; no
+synthetic conversation id is ever stored.
+
+Revision ID: 8b2f5c31d740
+Revises: 4e1c7a9d2b05
+Create Date: 2026-07-20 10:30:00.000000
+
+"""
+
+import logging
+from typing import Any, List, Sequence, Tuple, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "8b2f5c31d740"
+down_revision: Union[str, None] = "4e1c7a9d2b05"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+logger = logging.getLogger("alembic.runtime.migration")
+
+# Frozen copies of the importer's speaker roles: this migration must keep matching
+# the data as it was written, even if the application constants later change.
+_CUSTOMER_SPEAKERS = {"customer", "user"}
+_AGENT_SPEAKERS = {"agent", "assistant", "bot"}
+
+_IMPORTED_CASES = sa.text(
+    """
+    SELECT id,
+           input_data->>'message' AS message,
+           expected_output->>'value' AS answer
+    FROM test_cases
+    WHERE suite_id = :suite_id
+      AND is_deleted = 0
+      AND tags @> '["imported"]'::jsonb
+    ORDER BY created_at, id
+    """
+)
+
+# Live conversations whose live messages include this text from a customer.
+_CANDIDATE_CONVERSATIONS = sa.text(
+    """
+    SELECT DISTINCT message.conversation_id
+    FROM transcript_messages AS message
+    JOIN conversations AS conversation
+      ON conversation.id = message.conversation_id
+     AND conversation.is_deleted = 0
+    WHERE message.is_deleted = 0
+      AND message.text = :question
+      AND lower(message.speaker) = ANY(:customer_speakers)
+    """
+)
+
+_CONVERSATION_MESSAGES = sa.text(
+    """
+    SELECT speaker, text
+    FROM transcript_messages
+    WHERE conversation_id = :conversation_id
+      AND is_deleted = 0
+    ORDER BY sequence_number
+    """
+)
+
+
+def _exchanges_of(bind, conversation_id: Any) -> List[Tuple[str, str]]:
+    """Rebuild a conversation's ordered question/answer pairs, as the importer did."""
+    rows = bind.execute(
+        _CONVERSATION_MESSAGES, {"conversation_id": conversation_id}
+    ).all()
+
+    exchanges: List[Tuple[str, str]] = []
+    pending_question: str | None = None
+    for row in rows:
+        speaker = (row.speaker or "").lower()
+        if speaker in _CUSTOMER_SPEAKERS:
+            pending_question = row.text
+        elif speaker in _AGENT_SPEAKERS and pending_question is not None:
+            exchanges.append((pending_question, row.text))
+            pending_question = None
+    return exchanges
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+
+    suite_ids = bind.execute(
+        sa.text(
+            """
+            SELECT DISTINCT suite_id
+            FROM test_cases
+            WHERE is_deleted = 0
+              AND source_conversation_id IS NULL
+              AND tags @> '["imported"]'::jsonb
+            """
+        )
+    ).scalars().all()
+
+    linked = 0
+    skipped: list[tuple[str, str]] = []
+
+    for suite_id in suite_ids:
+        cases = bind.execute(_IMPORTED_CASES, {"suite_id": suite_id}).all()
+        cluster = [(case.message, case.answer) for case in cases]
+
+        if not cluster or not all(question and answer for question, answer in cluster):
+            skipped.append((str(suite_id), "cluster has an incomplete exchange"))
+            continue
+
+        candidate_ids = bind.execute(
+            _CANDIDATE_CONVERSATIONS,
+            {"question": cluster[0][0], "customer_speakers": list(_CUSTOMER_SPEAKERS)},
+        ).scalars().all()
+
+        matches = [
+            candidate_id
+            for candidate_id in candidate_ids
+            if _exchanges_of(bind, candidate_id) == cluster
+        ]
+
+        if len(matches) != 1:
+            skipped.append(
+                (
+                    str(suite_id),
+                    f"{len(matches)} conversations reproduce the cluster exactly",
+                )
+            )
+            continue
+
+        # Turn order comes from the transcript rebuild, not from row creation order.
+        for turn_index, case in enumerate(cases):
+            bind.execute(
+                sa.text(
+                    """
+                    UPDATE test_cases
+                    SET source_conversation_id = :conversation_id, turn_index = :turn_index
+                    WHERE id = :case_id
+                    """
+                ),
+                {
+                    "conversation_id": matches[0],
+                    "turn_index": turn_index,
+                    "case_id": case.id,
+                },
+            )
+        linked += len(cases)
+
+    logger.info("Linked %s imported cases to their source conversation", linked)
+    for suite_id, reason in skipped:
+        logger.warning(
+            "Suite %s left unlinked for manual review: %s", suite_id, reason
+        )
+
+
+def downgrade() -> None:
+    """Deliberately a no-op.
+
+    Clearing the columns cannot distinguish values recovered here from those the
+    importer set for real imports, and the latter are unrecoverable. The preceding
+    migration drops both columns, so nothing is left behind either way.
+    """

backend/alembic/versions/00097_unique_active_conversation_turns.py

@@ -0,0 +1,39 @@
+"""unique_active_conversation_turns
+
+Guards against a conversation being imported twice into the same dataset, which
+would produce duplicate turn positions and replay the conversation out of order.
+
+Only active rows are constrained. Manual and legacy cases keep a null
+source_conversation_id and are unaffected, since nulls never conflict.
+
+Revision ID: c5d9e0a71b38
+Revises: 8b2f5c31d740
+Create Date: 2026-07-21 09:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "c5d9e0a71b38"
+down_revision: Union[str, None] = "8b2f5c31d740"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_INDEX_NAME = "uq_test_cases_active_conversation_turn"
+
+
+def upgrade() -> None:
+    op.create_index(
+        _INDEX_NAME,
+        "test_cases",
+        ["suite_id", "source_conversation_id", "turn_index"],
+        unique=True,
+        postgresql_where="is_deleted = 0 AND source_conversation_id IS NOT NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(_INDEX_NAME, table_name="test_cases")

backend/alembic/versions/00098_add_status_to_test_results.py

@@ -0,0 +1,47 @@
+"""add_status_to_test_results
+
+Records why a case did or did not produce metrics, so skipped cases, execution
+failures and scoring failures can be told apart instead of all reading as
+"no metrics".
+
+Legacy rows that carry metrics are marked scored. Legacy rows without metrics
+keep a null status: the stored data cannot prove whether they were skipped,
+execution failures or scoring failures, and the reader treats null as unscored.
+
+Revision ID: e7a4b0c95d61
+Revises: c5d9e0a71b38
+Create Date: 2026-07-21 16:30:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "e7a4b0c95d61"
+down_revision: Union[str, None] = "c5d9e0a71b38"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "test_results",
+        sa.Column("status", sa.String(32), nullable=True),
+    )
+    # Only rows holding real metrics can be proven to have scored.
+    op.execute(
+        """
+        UPDATE test_results
+        SET status = 'scored'
+        WHERE metrics IS NOT NULL
+          AND jsonb_typeof(metrics) = 'object'
+          AND metrics <> '{}'::jsonb
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("test_results", "status")