feat(replay-vision): lead observations with a condensed prompt question

frontend/src/lib/constants.tsx

@@ -493,7 +493,6 @@ export const FEATURE_FLAGS = {
     REPLAY_UI_REDESIGN_2026: 'replay-ui-redesign-2026', // owner: #team-replay, New UI layout for replay
     REPLAY_VISION_ANALYSIS_NUDGE: 'replay-vision-analysis-nudge', // owner: #team-replay, in-player nudge offering an AI-drafted scanner after analyzing several recordings
     REPLAY_VISION_CALIBRATION_ACTIVATION: 'replay-vision-calibration-activation', // owner: #team-replay multivariate=control,badge,prompt, points a never-rated scanner at its Calibration tab
-    REPLAY_VISION_CALIBRATION_ENTRY_POINT: 'replay-vision-calibration-entry-point', // owner: #team-replay multivariate=control,test, links from a rated observation into the scanner's Calibration tab
     REPLAY_VISION_CALIBRATION_FEEDBACK_PROMPT: 'replay-vision-calibration-feedback-prompt', // owner: #team-replay multivariate=control,test, asks what the scanner should have concluded on a thumbs down
     REPLAY_VISION_CALIBRATION_TEST_NUDGE: 'replay-vision-calibration-test-nudge', // owner: #team-replay multivariate=control,test, prompts the user to test a recommendation before applying it
     REPLAY_VISION_HOME_REDESIGN_EXPERIMENT: 'replay-vision-home-redesign-experiment', // owner: #team-replay multivariate=control,test — gate on === 'test'; a truthy check turns on for control too

posthog/models/activity_logging/activity_log.py

@@ -451,6 +451,8 @@
     "search_suggestions_watermark",
     "search_suggestions_generated_at",
     "search_last_viewed_at",
+    "prompt_question",
+    "prompt_question_source",
     "limit_notified_period_start",
     "admission_budget_used",
     "admission_budget_refreshed_at",
@@ -1070,7 +1072,7 @@
 def changes_between(
     model_type: AuditableScope,
     previous: Optional[models.Model],
     current: Optional[models.Model],
 ) -> list[Change]:
     """
     Identifies changes between two models by comparing fields.
@@ -1267,7 +1269,7 @@
     user: Optional["User"],
     item_id: Optional[Union[int, str, UUID]],
     scope: str,
     activity: str,
     detail: Detail,
     was_impersonated: bool,
     client: Optional[str] = None,

products/replay_vision/backend/alert_destinations.py

@@ -19,6 +19,7 @@
     "alert_name": "{event.properties.alert_name}",
     "scanner_id": "{event.properties.scanner_id}",
     "scanner_name": "{event.properties.scanner_name}",
+    "scanner_question": "{event.properties.scanner_question}",
     "metric": "{event.properties.metric}",
     "metric_value": "{event.properties.metric_value}",
     "threshold": "{event.properties.threshold}",
@@ -38,12 +39,17 @@
 }
 
 
+# First, so a reader knows what was asked before reading the count or the matches it answers.
+_QUESTION_DETAIL = ("Question", "{event.properties.scanner_question_mrkdwn}")
+
+
 EVENT_KIND_CONFIG: dict[EventKind, EventKindSpec] = {
     "firing": EventKindSpec(
         event_id="$replay_vision_alert_firing",
         display_kind="firing",
         header="🔴 Replay vision alert '{event.properties.alert_name}' is firing",
         details=(
+            _QUESTION_DETAIL,
             (
                 "Threshold breached",
                 "{event.properties.metric_label} is {event.properties.metric_value} over the last "
@@ -65,6 +71,7 @@
         display_kind="resolved",
         header="🟢 Replay vision alert '{event.properties.alert_name}' has resolved",
         details=(
+            _QUESTION_DETAIL,
             (
                 "Current value",
                 "{event.properties.metric_label} is {event.properties.metric_value} over the last "
@@ -127,7 +134,7 @@
         event_id="$replay_vision_alert_match",
         display_kind="match",
         header="🔔 {event.properties.matched_count} new matching observations for '{event.properties.alert_name}'",
-        details=(("Matches", "{event.properties.summary}"),),
+        details=(_QUESTION_DETAIL, ("Matches", "{event.properties.summary}")),
         primary_action_url=_OBSERVATIONS_URL,
         primary_action_label="View observations",
         webhook_body={
@@ -139,6 +146,7 @@
                 "alert_name": "{event.properties.alert_name}",
                 "scanner_id": "{event.properties.scanner_id}",
                 "scanner_name": "{event.properties.scanner_name}",
+                "scanner_question": "{event.properties.scanner_question}",
                 "matched_count": "{event.properties.matched_count}",
                 "summary": "{event.properties.summary_text}",
                 "observation_ids": "{event.properties.observation_ids}",

products/replay_vision/backend/api/observations.py

@@ -64,6 +64,7 @@
 from products.replay_vision.backend.models.replay_observation_view import ReplayObservationView
 from products.replay_vision.backend.models.replay_scanner import ReplayScanner, ScannerOrigin, ScannerType
 from products.replay_vision.backend.observation_formatting import summarize_observation
+from products.replay_vision.backend.prompt_questions import question_for_snapshot
 from products.replay_vision.backend.scanner_access import (
     accessible_observations,
     can_read_targeted_experiment,
@@ -358,6 +359,23 @@ def get_media(self, obj: ReplayObservation) -> list[dict]:
             if media.asset.content_location
         ]
 
+    prompt_question = serializers.SerializerMethodField(
+        help_text=(
+            "The scanner's prompt condensed into the one question it answers about a session. Null when the "
+            "prompt has changed since this observation was scanned, since the question then describes a "
+            "different prompt; read `scanner_snapshot.scanner_config.prompt` instead."
+        ),
+    )
+
+    @extend_schema_field(serializers.CharField(allow_null=True))
+    def get_prompt_question(self, obj: ReplayObservation) -> str | None:
+        # Annotated by `hydrate_for_serialization`; a queryset that skipped it just has no question.
+        return question_for_snapshot(
+            snapshot_config=(obj.scanner_snapshot or {}).get("scanner_config"),
+            question=getattr(obj, "scanner_prompt_question", "") or "",
+            source=getattr(obj, "scanner_prompt_question_source", "") or "",
+        )
+
     summary_line = serializers.SerializerMethodField(
         help_text=(
             "One line of plain text saying what the scanner found: its verdict, score, tags or title, then its "
@@ -383,6 +401,7 @@ class Meta:
             "workflow_id",
             "scanner_snapshot",
             "scanner_result",
+            "prompt_question",
             "triggered_by",
             "triggered_by_user",
             "backfill_id",

products/replay_vision/backend/api/prompt_suggestions.py

@@ -43,6 +43,7 @@
     evaluation_in_flight,
     evaluation_supported,
 )
+from products.replay_vision.backend.prompt_questions import question_fields_for_save
 from products.replay_vision.backend.prompt_suggestions import (
     PromptSuggestionError,
     generate_prompt_suggestion,
@@ -423,6 +424,19 @@
             suggestion.applied_at = timezone.now()
             suggestion.applied_by = cast(User, request.user)
             suggestion.save(update_fields=["status", "applied_at", "applied_by"])
+        # A model call, so it waits until the row locks above are released. Conditional on the version this apply
+        # saved, so an edit that lands meanwhile keeps its own question.
+        question = question_fields_for_save(
+            team_id=self.team_id,
+            scanner_type=scanner.scanner_type,
+            scanner_config=config,
+            current_source=scanner.prompt_question_source,
+        )
+        if question:
+            ReplayScanner.objects.filter(pk=scanner.pk, scanner_version=scanner.scanner_version).update(
+                prompt_question=question["prompt_question"],
+                prompt_question_source=question["prompt_question_source"],
+            )
         user = cast(User, request.user)
         properties = {
             **_suggestion_properties(suggestion),
@@ -466,7 +480,7 @@
         ),
     )
     @action(detail=True, methods=["post"], required_scopes=["replay_scanner:write", "session_recording:read"])
     def evaluate(self, request: Request, **kwargs: Any) -> Response:
         refuse_scout_scanner_scan(is_scout_sandbox_request(request))
         scanner = self._scanner_for_url()
         self._require_editor(scanner)

products/replay_vision/backend/api/scanners.py

@@ -112,6 +112,7 @@
     ScannerType,
     apply_experiment_targeting,
 )
+from products.replay_vision.backend.prompt_questions import question_fields_for_save, scanner_question
 from products.replay_vision.backend.queries import (
     ESTIMATE_STALE_AFTER,
     MIN_SAMPLING_RATE,
@@ -490,6 +491,12 @@ class ReplayScannerSerializer(TaggedItemSerializerMixin, UserAccessControlSerial
             "classifiers add `tags`, scorers add `scale`, summarizers add optional `length`."
         ),
     )
+    prompt_question = serializers.SerializerMethodField(
+        help_text=(
+            "The current prompt condensed by AI into the one question the scanner answers about a session. "
+            "Falls back to the prompt's first line when no question matches the current prompt."
+        ),
+    )
     query = extend_schema_field(RecordingsQuery)(  # type: ignore[arg-type, type-var]
         serializers.JSONField(
             required=False,
@@ -638,6 +645,7 @@ class Meta:
             "scanner_type",
             "creation_method",
             "scanner_config",
+            "prompt_question",
             "query",
             "sampling_rate",
             "sampling_mode",
@@ -666,6 +674,7 @@ class Meta:
         ]
         read_only_fields = [
             "id",
+            "prompt_question",
             "scanner_version",
             "estimated_monthly_observations",
             "estimated_at",
@@ -684,6 +693,10 @@ class Meta:
             "user_access_level",
         ]
 
+    @extend_schema_field(serializers.CharField())
+    def get_prompt_question(self, scanner: ReplayScanner) -> str:
+        return scanner_question(scanner)
+
     @extend_schema_field(serializers.IntegerField())
     def get_credits_per_observation(self, scanner: ReplayScanner) -> int:
         return observation_credits_for_model(scanner.model)
@@ -893,6 +906,14 @@ def create(self, validated_data: dict[str, Any]) -> ReplayScanner:
         tags = validated_data.pop("tags", None)
         # Telemetry only, so it must not reach the model constructor.
         creation_method = validated_data.pop("creation_method", None)
+        # A model call, so it runs before the transaction opens.
+        validated_data.update(
+            question_fields_for_save(
+                team_id=team.id,
+                scanner_type=validated_data["scanner_type"],
+                scanner_config=validated_data.get("scanner_config", {}),
+            )
+        )
         # One transaction so a failed tag write can't leave an untagged scanner behind. Side effects stay outside.
         with transaction.atomic():
             try:
@@ -936,6 +957,16 @@ def update(self, instance: ReplayScanner, validated_data: dict[str, Any]) -> Rep
         before = {field: getattr(instance, field) for field in validated_data}
         was_enabled = instance.enabled
         limit_changed = "credit_limit" in validated_data and validated_data["credit_limit"] != instance.credit_limit
+        # After `before`, so the question is not reported as an edit. A model call, so before the transaction.
+        if "scanner_config" in validated_data:
+            validated_data.update(
+                question_fields_for_save(
+                    team_id=instance.team_id,
+                    scanner_type=validated_data.get("scanner_type", instance.scanner_type),
+                    scanner_config=validated_data["scanner_config"],
+                    current_source=instance.prompt_question_source,
+                )
+            )
         # One transaction so a failed tag write can't leave the columns updated with stale tags. Side effects stay outside.
         with transaction.atomic():
             try:
@@ -2134,6 +2165,9 @@ def duplicate(self, request: Request, **kwargs: Any) -> Response:
                     description=source.description,
                     scanner_type=source.scanner_type,
                     scanner_config=source.scanner_config,
+                    # Same prompt, so the source's question still describes it.
+                    prompt_question=source.prompt_question,
+                    prompt_question_source=source.prompt_question_source,
                     query=source.query,
                     sampling_rate=source.sampling_rate,
                     sampling_mode=source.sampling_mode,

products/replay_vision/backend/inline_scan.py

@@ -18,6 +18,7 @@
 
 from products.replay_vision.backend.fingerprint import config_fingerprint
 from products.replay_vision.backend.models.replay_scanner import ReplayScanner, ScannerOrigin, ScannerType
+from products.replay_vision.backend.prompt_questions import condense_prompt
 
 
 def inline_scan_key(*, scanner_type: str, scanner_config: dict[str, Any], model: str) -> str:
@@ -50,6 +51,7 @@ def create_inline_scanner(
     moment, and it wraps the losing INSERT in a savepoint so the unique violation doesn't poison an
     enclosing transaction.
     """
+    question = condense_prompt(team_id=team.id, scanner_type=scanner_type, scanner_config=scanner_config)
     scanner, _ = ReplayScanner.all_origins.get_or_create(
         team=team,
         origin=ScannerOrigin.INLINE,
@@ -64,6 +66,7 @@ def create_inline_scanner(
             "created_by": None,
             "scanner_type": scanner_type,
             "scanner_config": scanner_config,
+            **question.as_fields(),
             "model": model,
             # Nothing to sweep: no query, and disabled, which is what actually gates scheduling.
             "enabled": False,

products/replay_vision/backend/management/commands/backfill_replay_scanner_prompt_questions.py

@@ -0,0 +1,27 @@
+from typing import Any
+
+from django.core.management.base import BaseCommand, CommandParser
+
+from products.replay_vision.backend.prompt_questions import backfill_prompt_questions
+
+
+class Command(BaseCommand):
+    help = "Condense each Replay Vision scanner's prompt into the question the observation page shows"
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        parser.add_argument("--team-id", type=int, default=None, help="Only this team's scanners")
+        parser.add_argument(
+            "--include-inline", action="store_true", help="Also cover inline scanners minted by one-off scans"
+        )
+        parser.add_argument("--limit", type=int, default=None, help="Stop after writing this many questions")
+        parser.add_argument("--dry-run", action="store_true", help="Count the scanners due a question, write nothing")
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        result = backfill_prompt_questions(
+            team_id=options["team_id"],
+            include_inline=bool(options["include_inline"]),
+            limit=options["limit"],
+            dry_run=bool(options["dry_run"]),
+        )
+        verb = "Would write" if options["dry_run"] else "Wrote"
+        self.stdout.write(f"Checked {result.checked} scanners. {verb} {result.written} questions.")

products/replay_vision/backend/migrations/0101_replayscanner_prompt_question.py

@@ -0,0 +1,33 @@
+# Generated by Django 5.2.17 on 2026-09-29 09:04
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("replay_vision", "0100_team_replay_vision_config"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="replayscanner",
+            name="prompt_question",
+            field=models.TextField(
+                blank=True,
+                db_default="",
+                default="",
+                help_text="The prompt condensed by AI into one question, shown above an observation's answer.",
+            ),
+        ),
+        migrations.AddField(
+            model_name="replayscanner",
+            name="prompt_question_source",
+            field=models.CharField(
+                blank=True,
+                db_default="",
+                default="",
+                help_text="`prompt_fingerprint` of the prompt `prompt_question` was condensed from. A mismatch means it is stale.",
+                max_length=64,
+            ),
+        ),
+    ]

products/replay_vision/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0100_team_replay_vision_config
+0101_replayscanner_prompt_question

products/replay_vision/backend/models/replay_observation.py

@@ -220,7 +220,12 @@ def hydrate_for_serialization(
                 queryset=ReplayObservationMedia.objects.unscoped().select_related("asset").order_by("kind", "position"),
             )
         )
-        .annotate(scanner_origin=F("scanner__origin"), viewed=viewed)
+        .annotate(
+            scanner_origin=F("scanner__origin"),
+            scanner_prompt_question=F("scanner__prompt_question"),
+            scanner_prompt_question_source=F("scanner__prompt_question_source"),
+            viewed=viewed,
+        )
     )
 
 

products/replay_vision/backend/models/replay_scanner.py

@@ -1,3 +1,4 @@
+import hashlib
 import datetime as dt
 from typing import TYPE_CHECKING
 
@@ -87,6 +88,11 @@ class ScannerOrigin(models.TextChoices):
     INLINE = "inline", "Inline"
 
 
+def prompt_fingerprint(prompt: str) -> str:
+    """Identifies a prompt's text, so a condensed question can be matched to the prompt it came from."""
+    return hashlib.sha256(prompt.encode()).hexdigest()
+
+
 def initial_watermark() -> "datetime":
     """A new scanner's sweep watermark, started one settle-interval back so its first sweep immediately picks up
     recordings that have just cleared the settle window instead of a ~settle-interval cold start; it advances
@@ -277,6 +283,22 @@ class ReplayScanner(Taggable, ModelActivityMixin, UUIDModel):
         help_text="When the Search tab last asked for this scanner's suggestions. Only viewed scanners refresh.",
     )
 
+    # Written with the prompt by every path that sets one, see `prompt_questions`. Not version-tracked: it
+    # restates the prompt and changes nothing about how the scanner scans.
+    prompt_question = models.TextField(
+        blank=True,
+        default="",
+        db_default="",
+        help_text="The prompt condensed by AI into one question, shown above an observation's answer.",
+    )
+    prompt_question_source = models.CharField(
+        max_length=64,
+        blank=True,
+        default="",
+        db_default="",
+        help_text="`prompt_fingerprint` of the prompt `prompt_question` was condensed from. A mismatch means it is stale.",
+    )
+
     # Not "monthly": this resets with the org's billing period, which is only a calendar month
     # until billing syncs a real one. See quota.current_period_bounds.
     credit_limit = models.PositiveIntegerField(