feat(replays): Add Breadcrumb AI Summaries

src/sentry/api/urls.py

@@ -323,6 +323,9 @@
 from sentry.replays.endpoints.project_replay_recording_segment_index import (
     ProjectReplayRecordingSegmentIndexEndpoint,
 )
+from sentry.replays.endpoints.project_replay_summarize_breadcrumbs import (
+    ProjectReplaySummarizeBreadcrumbsEndpoint,
+)
 from sentry.replays.endpoints.project_replay_video_details import ProjectReplayVideoDetailsEndpoint
 from sentry.replays.endpoints.project_replay_viewed_by import ProjectReplayViewedByEndpoint
 from sentry.rules.history.endpoints.project_rule_group_history import (
@@ -2690,6 +2693,11 @@ def create_group_urls(name_prefix: str) -> list[URLPattern | URLResolver]:
         ProjectReplayRecordingSegmentIndexEndpoint.as_view(),
         name="sentry-api-0-project-replay-recording-segment-index",
     ),
+    re_path(
+        r"^(?P<organization_id_or_slug>[^/]+)/(?P<project_id_or_slug>[^\/]+)/replays/(?P<replay_id>[\w-]+)/summarize/breadcrumbs/$",
+        ProjectReplaySummarizeBreadcrumbsEndpoint.as_view(),
+        name="sentry-api-0-project-replay-summarize-breadcrumbs",
+    ),
     re_path(
         r"^(?P<organization_id_or_slug>[^/]+)/(?P<project_id_or_slug>[^\/]+)/replays/(?P<replay_id>[\w-]+)/recording-segments/(?P<segment_id>\d+)/$",
         ProjectReplayRecordingSegmentDetailsEndpoint.as_view(),

src/sentry/replays/blueprints/api.md

@@ -583,3 +583,34 @@ A POST request is issued with no body. The URL and authorization context is used
     Cookie: \_ga=GA1.2.17576183...
 
 - Response 204
+
+## Replay Summarize Breadcrumb [/projects/<organization_id_or_slug>/<project_id_or_slug>/replays/<replay_id>/summarize/breadcrumbs/]
+
+### Fetch Replay Breadcrumb Summary [GET]
+
+| Column                   | Type            | Description                                                                                   |
+| ------------------------ | --------------- | --------------------------------------------------------------------------------------------- |
+| title                    | str             | The main title of the user journey summary.                                                   |
+| summary                  | str             | A concise summary featuring the highlights of the user's journey while using the application. |
+| time_ranges              | list[TimeRange] | A list of TimeRange objects.                                                                  |
+| time_ranges.period_start | number          | The start time (UNIX timestamp) of the analysis window.                                       |
+| time_ranges.period_end   | number          | The end time (UNIX timestamp) of the analysis window.                                         |
+| time_ranges.period_title | str             | A concise summary utilizing 6 words or fewer describing what happened during the time range.  |
+
+- Response 200
+
+  ```json
+  {
+    "data": {
+      "title": "Something Happened",
+      "summary": "The application broke",
+      "time_ranges": [
+        {
+          "period_start": 1749584581.5356228,
+          "period_end": 1749584992.912,
+          "period_title": "Second Replay Load Failure"
+        }
+      ]
+    }
+  }
+  ```

src/sentry/replays/endpoints/project_replay_summarize_breadcrumbs.py

@@ -0,0 +1,96 @@
+import functools
+from collections.abc import Generator, Iterator
+from typing import Any
+
+import requests
+import sentry_sdk
+from django.conf import settings
+from drf_spectacular.utils import extend_schema
+from rest_framework.exceptions import ParseError
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import features
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import region_silo_endpoint
+from sentry.api.bases.project import ProjectEndpoint
+from sentry.api.paginator import GenericOffsetPaginator
+from sentry.replays.lib.storage import RecordingSegmentStorageMeta, storage
+from sentry.replays.usecases.ingest.event_parser import as_log_message
+from sentry.replays.usecases.reader import fetch_segments_metadata, iter_segment_data
+from sentry.seer.signed_seer_api import sign_with_seer_secret
+from sentry.utils import json
+
+
+@region_silo_endpoint
+@extend_schema(tags=["Replays"])
+class ProjectReplaySummarizeBreadcrumbsEndpoint(ProjectEndpoint):
+    owner = ApiOwner.REPLAY
+    publish_status = {
+        "GET": ApiPublishStatus.EXPERIMENTAL,
+    }
+
+    def __init__(self, **options) -> None:
+        storage.initialize_client()
+        super().__init__(**options)
+
+    def get(self, request: Request, project, replay_id: str) -> Response:
+        """Return a collection of replay recording segments."""
+        if (
+            not features.has(
+                "organizations:session-replay", project.organization, actor=request.user
+            )
+            or not features.has(
+                "organizations:replay-ai-summaries", project.organization, actor=request.user
+            )
+            or not features.has(
+                "organizations:gen-ai-features", project.organization, actor=request.user
+            )
+        ):
+            return self.respond(status=404)
+
+        return self.paginate(
+            request=request,
+            paginator_cls=GenericOffsetPaginator,
+            data_fn=functools.partial(fetch_segments_metadata, project.id, replay_id),
+            on_results=analyze_recording_segments,
+        )
+
+
+@sentry_sdk.trace
+def analyze_recording_segments(segments: list[RecordingSegmentStorageMeta]) -> dict[str, Any]:
+    request_data = json.dumps({"logs": get_request_data(iter_segment_data(segments))})
+
+    # XXX: I have to deserialize this request so it can be "automatically" reserialized by the
+    # paginate method. This is less than ideal.
+    return json.loads(make_seer_request(request_data).decode("utf-8"))
+
+
+def make_seer_request(request_data: str) -> bytes:
+    # XXX: Request isn't streaming. Limitation of Seer authentication. Would be much faster if we
+    # could stream the request data since the GCS download will (likely) dominate latency.
+    response = requests.post(
+        f"{settings.SEER_AUTOFIX_URL}/v1/automation/summarize/replay/breadcrumbs",
+        data=request_data,
+        headers={
+            "content-type": "application/json;charset=utf-8",
+            **sign_with_seer_secret(request_data.encode()),
+        },
+    )
+    if response.status_code != 200:
+        raise ParseError("A summary could not be produced at this time.")
+
+    return response.content
+
+
+def get_request_data(iterator: Iterator[tuple[int, memoryview]]) -> list[str]:
+    return list(gen_request_data(map(lambda r: r[1], iterator)))
+
+
+def gen_request_data(segments: Iterator[memoryview]) -> Generator[str]:
+    for segment in segments:
+        for event in json.loads(segment.tobytes().decode("utf-8")):
+            message = as_log_message(event)
+            if message:
+                yield message

src/sentry/replays/usecases/ingest/event_parser.py

@@ -2,7 +2,9 @@
 
 import random
 from dataclasses import dataclass
+from enum import Enum
 from typing import Any
+from urllib.parse import urlparse
 
 import sentry_sdk
 
@@ -228,3 +230,138 @@ def _get_breadcrumb_event(
         return MutationEvent(payload)
     else:
         return None
+
+
+class EventType(Enum):
+    CLICK = 0
+    DEAD_CLICK = 1
+    RAGE_CLICK = 2
+    NAVIGATION = 3
+    CONSOLE = 4
+    UI_BLUR = 5
+    UI_FOCUS = 6
+    RESOURCE_FETCH = 7
+    RESOURCE_XHR = 8
+    LCP = 9
+    FCP = 10
+    HYDRATION_ERROR = 11
+    MUTATIONS = 12
+    UNKNOWN = 13
+    FEEDBACK = 14
+
+
+def which(event: dict[str, Any]) -> EventType:
+    """Identify the passed event.
+
+    This function helpfully hides the dirty data munging necessary to identify an event type and
+    helpfully reduces the number of operations required by reusing context from previous
+    branches.
+    """
+    if event.get("type") == 5:
+        if event["data"]["tag"] == "breadcrumb":
+            payload = event["data"]["payload"]
+            category = payload["category"]
+            if category == "ui.click":
+                return EventType.CLICK
+            elif category == "ui.slowClickDetected":
+                is_timeout_reason = payload["data"].get("endReason") == "timeout"
+                is_target_tagname = payload["data"].get("node", {}).get("tagName") in (
+                    "a",
+                    "button",
+                    "input",
+                )
+                timeout = payload["data"].get("timeAfterClickMs", 0) or payload["data"].get(
+                    "timeafterclickms", 0
+                )
+
+                if is_timeout_reason and is_target_tagname and timeout >= 7000:
+                    is_rage = (
+                        payload["data"].get("clickCount", 0) or payload["data"].get("clickcount", 0)
+                    ) >= 5
+                    if is_rage:
+                        return EventType.RAGE_CLICK
+                    else:
+                        return EventType.DEAD_CLICK
+            elif category == "navigation":
+                return EventType.NAVIGATION
+            elif category == "console":
+                return EventType.CONSOLE
+            elif category == "ui.blur":
+                return EventType.UI_BLUR
+            elif category == "ui.focus":
+                return EventType.UI_FOCUS
+            elif category == "replay.hydrate-error":
+                return EventType.HYDRATION_ERROR
+            elif category == "replay.mutations":
+                return EventType.MUTATIONS
+            elif category == "sentry.feedback":
+                return EventType.FEEDBACK
+        elif event["data"]["tag"] == "performanceSpan":
+            payload = event["data"]["payload"]
+            op = payload["op"]
+            if op == "resource.fetch":
+                return EventType.RESOURCE_FETCH
+            elif op == "resource.xhr":
+                return EventType.RESOURCE_XHR
+            elif op == "web-vital":
+                if payload["description"] == "largest-contentful-paint":
+                    return EventType.LCP
+                elif payload["description"] == "first-contentful-paint":
+                    return EventType.FCP
+    return EventType.UNKNOWN
+
+
+def as_log_message(event: dict[str, Any]) -> str | None:
+    """Return an event as a log message.
+
+    Useful in AI contexts where the event's structure is an impediment to the AI's understanding
+    of the interaction log. Not every event produces a log message. This function is overly coupled
+    to the AI use case. In later iterations, if more or all log messages are desired, this function
+    should be forked.
+    """
+    event_type = which(event)
+    timestamp = event.get("timestamp", 0.0)
+
+    match event_type:
+        case EventType.CLICK:
+            return f"User clicked on {event["data"]["payload"]["message"]} at {timestamp}"
+        case EventType.DEAD_CLICK:
+            return f"User clicked on {event["data"]["payload"]["message"]} but the triggered action was slow to complete at {timestamp}"
+        case EventType.RAGE_CLICK:
+            return f"User rage clicked on {event["data"]["payload"]["message"]} but the triggered action was slow to complete at {timestamp}"
+        case EventType.NAVIGATION:
+            return f"User navigated to: {event["data"]["payload"]["data"]["to"]} at {timestamp}"
+        case EventType.CONSOLE:
+            return f"Logged: {event["data"]["payload"]["message"]} at {timestamp}"
+        case EventType.UI_BLUR:
+            return f"User looked away from the tab at {timestamp}."
+        case EventType.UI_FOCUS:
+            return f"User returned to tab at {timestamp}."
+        case EventType.RESOURCE_FETCH:
+            payload = event["data"]["payload"]
+            parsed_url = urlparse(payload["description"])
+
+            path = f"{parsed_url.path}?{parsed_url.query}"
+            size = payload["data"]["response"]["size"]
+            status_code = payload["data"]["statusCode"]
+            duration = payload["endTimestamp"] - payload["startTimestamp"]
+            method = payload["data"]["method"]
+            return f'Application initiated request: "{method} {path} HTTP/2.0" {status_code} {size}; took {duration} milliseconds at {timestamp}'
+        case EventType.RESOURCE_XHR:
+            return None
+        case EventType.LCP:
+            duration = event["data"]["payload"]["data"]["size"]
+            rating = event["data"]["payload"]["data"]["rating"]
+            return f"Application largest contentful paint: {duration} ms and has a {rating} rating"
+        case EventType.FCP:
+            duration = event["data"]["payload"]["data"]["size"]
+            rating = event["data"]["payload"]["data"]["rating"]
+            return f"Application first contentful paint: {duration} ms and has a {rating} rating"
+        case EventType.HYDRATION_ERROR:
+            return f"There was a hydration error on the page at {timestamp}."
+        case EventType.MUTATIONS:
+            return None
+        case EventType.UNKNOWN:
+            return None
+        case EventType.FEEDBACK:
+            return "The user filled out a feedback form describing their experience using the application."

src/sentry/replays/usecases/reader.py

@@ -7,6 +7,7 @@
 from datetime import datetime, timedelta
 from typing import Any
 
+import sentry_sdk
 from django.conf import settings
 from django.db.models import Prefetch
 from snuba_sdk import (
@@ -39,6 +40,7 @@
 # METADATA QUERY BEHAVIOR.
 
 
+@sentry_sdk.trace
 def fetch_segments_metadata(
     project_id: int,
     replay_id: str,
@@ -72,6 +74,7 @@ def fetch_segment_metadata(
     return fetch_direct_storage_segment_meta(project_id, replay_id, segment_id)
 
 
+@sentry_sdk.trace
 def fetch_filestore_segments_meta(
     project_id: int,
     replay_id: str,
@@ -134,6 +137,7 @@ def fetch_filestore_segment_meta(
     )
 
 
+@sentry_sdk.trace
 def fetch_direct_storage_segments_meta(
     project_id: int,
     replay_id: str,
@@ -168,6 +172,7 @@ def fetch_direct_storage_segment_meta(
         return results[0]
 
 
+@sentry_sdk.trace
 def has_archived_segment(project_id: int, replay_id: str) -> bool:
     """Return true if an archive row exists for this replay."""
     snuba_request = Request(
@@ -250,21 +255,30 @@ def segment_row_to_storage_meta(
 # BLOB DOWNLOAD BEHAVIOR.
 
 
+@sentry_sdk.trace
 def download_segments(segments: list[RecordingSegmentStorageMeta]) -> Iterator[bytes]:
     """Download segment data from remote storage."""
     yield b"["
 
+    for i, segment in iter_segment_data(segments):
+        yield segment
+        if i < len(segments) - 1:
+            yield b","
+
+    yield b"]"
+
+
+def iter_segment_data(
+    segments: list[RecordingSegmentStorageMeta],
+) -> Iterator[tuple[int, memoryview]]:
     with ThreadPoolExecutor(max_workers=10) as pool:
         segment_data = pool.map(_download_segment, segments)
 
     for i, result in enumerate(segment_data):
         if result is None:
-            yield b"[]"
+            yield i, memoryview(b"[]")
         else:
-            yield result[1]
-        if i < len(segments) - 1:
-            yield b","
-    yield b"]"
+            yield i, result[1]
 
 
 def download_segment(segment: RecordingSegmentStorageMeta, span: Any) -> bytes:
@@ -287,7 +301,9 @@ def download_video(segment: RecordingSegmentStorageMeta) -> bytes | None:
         return video
 
 
-def _download_segment(segment: RecordingSegmentStorageMeta) -> tuple[bytes | None, bytes] | None:
+def _download_segment(
+    segment: RecordingSegmentStorageMeta,
+) -> tuple[memoryview | None, memoryview] | None:
     driver = filestore if segment.file_id else storage
 
     result = driver.get(segment)
@@ -298,6 +314,7 @@ def _download_segment(segment: RecordingSegmentStorageMeta) -> tuple[bytes | Non
     return unpack(decompressed)
 
 
+@sentry_sdk.trace
 def decompress(buffer: bytes) -> bytes:
     """Return decompressed output."""
     # If the file starts with a valid JSON character we assume its uncompressed.