MAINT: Ordering AttackResults by timestamp for perf

pyrit/backend/mappers/attack_mappers.py

@@ -231,22 +231,22 @@ def _resolve_summary_timestamps(ar: AttackResult) -> tuple[datetime, datetime]:
     """
     Resolve ``created_at`` / ``updated_at`` for a summary.
 
-    Resolution order for ``created_at``: explicit metadata override, then the
-    persisted ``AttackResult.timestamp``, and finally ``datetime.now`` as a
-    last-resort fallback for never-persisted results.
+    ``updated_at`` is the persisted ``AttackResult.timestamp`` — the single indexed recency
+    key that manual edits (add message, branch, promote conversation) bump. ``created_at``
+    comes from the display-only ``metadata.created_at`` override, falling back to
+    ``AttackResult.timestamp`` and finally ``datetime.now`` for never-persisted results.
 
     Returns:
         A ``(created_at, updated_at)`` tuple.
     """
     created_str = ar.metadata.get("created_at")
-    updated_str = ar.metadata.get("updated_at")
     if created_str:
         created_at = datetime.fromisoformat(created_str)
     elif ar.timestamp is not None:
         created_at = ar.timestamp
     else:
         created_at = datetime.now(timezone.utc)
-    updated_at = datetime.fromisoformat(updated_str) if updated_str else created_at
+    updated_at = ar.timestamp if ar.timestamp is not None else created_at
     return created_at, updated_at
 
 

pyrit/backend/services/attack_service.py

@@ -358,9 +358,9 @@ async def create_attack_async(self, *, request: CreateAttackRequest) -> CreateAt
                 ),
             ),
             outcome=AttackOutcome.UNDETERMINED,
+            timestamp=now,
             metadata={
                 "created_at": now.isoformat(),
-                "updated_at": now.isoformat(),
             },
             labels=labels,
         )
@@ -414,15 +414,11 @@ async def update_attack_async(self, *, attack_result_id: str, request: UpdateAtt
         }
         new_outcome = outcome_map.get(request.outcome, AttackOutcome.UNDETERMINED)
 
-        ar = results[0]
-        updated_metadata = dict(ar.metadata) if ar.metadata else {}
-        updated_metadata["updated_at"] = datetime.now(timezone.utc).isoformat()
-
         self._memory.update_attack_result_by_id(
             attack_result_id=attack_result_id,
             update_fields={
                 "outcome": new_outcome.value,
-                "attack_metadata": updated_metadata,
+                "timestamp": datetime.now(timezone.utc),
             },
         )
 
@@ -527,14 +523,11 @@ async def create_related_conversation_async(
         # Add to pruned_conversation_ids so user-created branches are visible in the GUI history panel.
         existing_pruned = ar.get_pruned_conversation_ids()
 
-        updated_metadata = dict(ar.metadata or {})
-        updated_metadata["updated_at"] = now.isoformat()
-
         self._memory.update_attack_result_by_id(
             attack_result_id=attack_result_id,
             update_fields={
                 "pruned_conversation_ids": existing_pruned + [new_conversation_id],
-                "attack_metadata": updated_metadata,
+                "timestamp": now,
             },
         )
 
@@ -590,16 +583,14 @@ async def update_main_conversation_async(
         updated_pruned.append(ar.conversation_id)
 
         now = datetime.now(timezone.utc)
-        updated_metadata = dict(ar.metadata or {})
-        updated_metadata["updated_at"] = now.isoformat()
 
         self._memory.update_attack_result_by_id(
             attack_result_id=attack_result_id,
             update_fields={
                 "conversation_id": target_conv_id,
                 "pruned_conversation_ids": updated_pruned if updated_pruned else None,
                 "adversarial_chat_conversation_ids": updated_adversarial if updated_adversarial else None,
-                "attack_metadata": updated_metadata,
+                "timestamp": now,
             },
         )
 
@@ -752,12 +743,12 @@ async def _update_attack_after_message_async(
         self, *, attack_result_id: str, ar: AttackResult, request: AddMessageRequest
     ) -> None:
         """
-        Update attack metadata and converter tracking after a message is added.
-        """
-        updated_metadata = dict(ar.metadata or {})
-        updated_metadata["updated_at"] = datetime.now(timezone.utc).isoformat()
+        Update attack recency and converter tracking after a message is added.
 
-        update_fields: dict[str, Any] = {"attack_metadata": updated_metadata}
+        Bumps the attack's ``timestamp`` column (the single indexed recency key) so the edited
+        conversation re-floats to the top of the History view.
+        """
+        update_fields: dict[str, Any] = {"timestamp": datetime.now(timezone.utc)}
 
         if request.converter_ids:
             converter_objs = get_converter_service().get_converter_objects_for_ids(converter_ids=request.converter_ids)
@@ -924,18 +915,16 @@ def _encode_attack_cursor(*, cursor: AttackResultsKeysetCursor, fingerprint: str
         """
         Encode a keyset anchor and its filter fingerprint into an opaque pagination cursor.
 
-        The anchor's recency string and timestamp can contain ``.``/``:``/``-`` (ISO
-        timestamps), so the payload is JSON-serialized and base64url-encoded rather than
-        joined with a delimiter, keeping the cursor an unambiguous opaque token for
-        ``_decode_attack_cursor``.
+        The anchor's timestamp can contain ``.``/``:``/``-`` (ISO timestamps), so the payload
+        is JSON-serialized and base64url-encoded rather than joined with a delimiter, keeping
+        the cursor an unambiguous opaque token for ``_decode_attack_cursor``.
 
         Returns:
-            An opaque base64url cursor string encoding ``{fingerprint, recency, timestamp,
+            An opaque base64url cursor string encoding ``{fingerprint, timestamp,
             attack_result_id}``.
         """
         payload = {
             "f": fingerprint,
-            "r": cursor.recency,
             "t": cursor.timestamp.isoformat(),
             "i": cursor.attack_result_id,
         }
@@ -947,12 +936,12 @@ def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResu
         """
         Decode the opaque list-attacks cursor into a keyset (seek) anchor.
 
-        The cursor encodes the previous page's last-row recency anchor together with a
+        The cursor encodes the previous page's last-row timestamp anchor together with a
         fingerprint of the filter set it was generated for (see ``_attack_filter_fingerprint``).
         A cursor is honored only when its fingerprint matches the current request's filters;
-        malformed, legacy (offset/attack-result-id), or filter-mismatched cursors fall back to
-        the first page (``None``) so a stale cursor degrades gracefully instead of raising or
-        seeking within the wrong result set.
+        malformed, legacy (offset/attack-result-id/recency-string), or filter-mismatched cursors
+        fall back to the first page (``None``) so a stale cursor degrades gracefully instead of
+        raising or seeking within the wrong result set.
 
         Returns:
             The decoded ``AttackResultsKeysetCursor``, or ``None`` to start at the first page.
@@ -966,10 +955,9 @@ def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResu
             return None
         if not isinstance(payload, dict) or payload.get("f") != fingerprint:
             return None
-        recency = payload.get("r")
         raw_timestamp = payload.get("t")
         attack_result_id = payload.get("i")
-        if not isinstance(recency, str) or not isinstance(raw_timestamp, str) or not isinstance(attack_result_id, str):
+        if not isinstance(raw_timestamp, str) or not isinstance(attack_result_id, str):
             return None
         try:
             timestamp = datetime.fromisoformat(raw_timestamp)
@@ -989,7 +977,7 @@ def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResu
             # A crafted cursor near datetime's min/max with a large UTC offset overflows the
             # representable range when shifted to UTC; treat it as malformed and restart at page one.
             return None
-        return AttackResultsKeysetCursor(recency=recency, timestamp=timestamp, attack_result_id=attack_result_id)
+        return AttackResultsKeysetCursor(timestamp=timestamp, attack_result_id=attack_result_id)
 
     # ========================================================================
     # Private Helper Methods - Duplicate / Branch

pyrit/memory/alembic/versions/d7e9f1a3b5c6_index_attack_result_recency.py

@@ -0,0 +1,190 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Index AttackResultEntries for the History recency query and make the ``timestamp``
+column the single source of truth for "last updated".
+
+Adds an index on ``conversation_id`` (serves the per-conversation dedup window) and a
+composite ``(timestamp, id)`` index (serves the recency ORDER BY + keyset seek). Backfills
+``timestamp`` from the legacy ``attack_metadata.updated_at``/``created_at`` JSON keys so
+manually-edited conversations keep their current History order, and drops the now-redundant
+``updated_at`` key from the JSON metadata.
+
+Revision ID: d7e9f1a3b5c6
+Revises: 3f6e8a0c2d4b
+Create Date: 2026-07-20 12:00:00.000000
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence  # noqa: TC003
+from datetime import datetime
+
+import sqlalchemy as sa
+from alembic import op
+
+from pyrit.memory.memory_models import CustomUUID, UTCDateTime
+
+# revision identifiers, used by Alembic.
+revision: str = "d7e9f1a3b5c6"
+down_revision: str | Sequence[str] | None = "3f6e8a0c2d4b"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+logger = logging.getLogger(__name__)
+
+# Bounded batch size for the recency backfill updates. Grouping the row updates into a single
+# executemany per batch keeps the number of statements (and, on SQL Server, database roundtrips)
+# proportional to the row count / batch size instead of one statement per row.
+_BACKFILL_UPDATE_BATCH_SIZE = 400
+
+
+def _attack_results_table() -> sa.Table:
+    """
+    Build a lightweight typed table clause for reading/writing the columns this migration
+    touches, so ``UTCDateTime``/``JSON`` bind and result processors run on both backends.
+
+    Returns:
+        sa.Table: A minimal ``AttackResultEntries`` table with the id, attack_metadata, and
+        timestamp columns.
+    """
+    metadata_obj = sa.MetaData()
+    return sa.Table(
+        "AttackResultEntries",
+        metadata_obj,
+        sa.Column("id", CustomUUID(), primary_key=True),
+        sa.Column("attack_metadata", sa.JSON()),
+        sa.Column("timestamp", UTCDateTime()),
+        extend_existing=True,
+    )
+
+
+def _parse_iso(value: object) -> datetime | None:
+    """
+    Parse an ISO-8601 string (tolerating a trailing ``Z``) into a datetime, or ``None``.
+
+    Returns:
+        datetime | None: The parsed datetime, or ``None`` when ``value`` is not a parseable
+        ISO string.
+    """
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def upgrade() -> None:
+    """Apply this schema upgrade."""
+    op.create_index(
+        "ix_AttackResultEntries_conversation_id",
+        "AttackResultEntries",
+        ["conversation_id"],
+    )
+    op.create_index(
+        "ix_AttackResultEntries_timestamp_id",
+        "AttackResultEntries",
+        ["timestamp", "id"],
+    )
+    _backfill_timestamp_from_metadata()
+
+
+def downgrade() -> None:
+    """Revert this schema upgrade."""
+    # Restore metadata.updated_at from the timestamp column so the legacy JSON recency sort
+    # reflects each row's last-updated time again. The original created/updated split cannot be
+    # perfectly recovered (it was collapsed into the single timestamp), so updated_at is set to
+    # the current timestamp value.
+    _restore_updated_at_to_metadata()
+    op.drop_index("ix_AttackResultEntries_timestamp_id", table_name="AttackResultEntries")
+    op.drop_index("ix_AttackResultEntries_conversation_id", table_name="AttackResultEntries")
+
+
+def _backfill_timestamp_from_metadata() -> None:
+    """
+    Set ``timestamp = COALESCE(parse(metadata.updated_at), parse(metadata.created_at),
+    timestamp)`` for existing rows and drop the now-redundant ``updated_at`` JSON key.
+
+    Only rows that carried a JSON recency key change; programmatic rows already have
+    ``timestamp`` equal to their creation time, and rows with no ``attack_metadata`` at all are
+    skipped by the query filter since they can never carry a recency key. The row updates are
+    applied in bounded ``executemany`` batches rather than one statement per row. Idempotent:
+    rerunning is a no-op once the ``updated_at`` key is gone and ``timestamp`` already reflects it.
+    """
+    bind = op.get_bind()
+    table = _attack_results_table()
+
+    rows = bind.execute(
+        sa.select(table.c.id, table.c.attack_metadata, table.c.timestamp).where(table.c.attack_metadata.isnot(None))
+    ).fetchall()
+
+    updates: list[dict[str, object]] = []
+    for row in rows:
+        metadata = dict(row.attack_metadata) if isinstance(row.attack_metadata, dict) else {}
+        new_timestamp = _parse_iso(metadata.get("updated_at")) or _parse_iso(metadata.get("created_at"))
+
+        had_updated_at = "updated_at" in metadata
+        new_metadata = {k: v for k, v in metadata.items() if k != "updated_at"}
+
+        timestamp_changes = new_timestamp is not None and new_timestamp != row.timestamp
+        if not timestamp_changes and not had_updated_at:
+            continue
+
+        # Always carry a timestamp so a single uniform executemany can update both columns;
+        # rows that only strip the JSON key rewrite their existing timestamp unchanged.
+        updates.append(
+            {
+                "b_id": row.id,
+                "b_metadata": new_metadata or None,
+                "b_timestamp": new_timestamp if timestamp_changes else row.timestamp,
+            }
+        )
+
+    _apply_backfill_updates(bind=bind, table=table, updates=updates)
+
+    if updates:
+        logger.info(f"Attack recency backfill: updated {len(updates)} AttackResultEntries row(s).")
+
+
+def _restore_updated_at_to_metadata() -> None:
+    """
+    Downgrade inverse: write ``metadata.updated_at = timestamp.isoformat()`` for every row so
+    the legacy JSON recency expression orders rows by their last-updated time again. The row
+    updates are applied in bounded ``executemany`` batches rather than one statement per row.
+    """
+    bind = op.get_bind()
+    table = _attack_results_table()
+
+    rows = bind.execute(
+        sa.select(table.c.id, table.c.attack_metadata, table.c.timestamp).where(table.c.timestamp.isnot(None))
+    ).fetchall()
+
+    updates: list[dict[str, object]] = []
+    for row in rows:
+        metadata = dict(row.attack_metadata) if isinstance(row.attack_metadata, dict) else {}
+        metadata["updated_at"] = row.timestamp.isoformat()
+        updates.append({"b_id": row.id, "b_metadata": metadata, "b_timestamp": row.timestamp})
+
+    _apply_backfill_updates(bind=bind, table=table, updates=updates)
+
+
+def _apply_backfill_updates(*, bind: sa.engine.Connection, table: sa.Table, updates: list[dict[str, object]]) -> None:
+    """
+    Apply keyed ``(b_id, b_metadata, b_timestamp)`` updates to ``AttackResultEntries`` in bounded
+    ``executemany`` batches. Each batch is a single compiled UPDATE executed against a slice of the
+    parameter list, so the statement count scales with ``len(updates) / _BACKFILL_UPDATE_BATCH_SIZE``.
+    """
+    if not updates:
+        return
+
+    update_stmt = (
+        sa.update(table)
+        .where(table.c.id == sa.bindparam("b_id"))
+        .values(attack_metadata=sa.bindparam("b_metadata"), timestamp=sa.bindparam("b_timestamp"))
+    )
+    for start in range(0, len(updates), _BACKFILL_UPDATE_BATCH_SIZE):
+        bind.execute(update_stmt, updates[start : start + _BACKFILL_UPDATE_BATCH_SIZE])