diff --git a/products/replay_vision/backend/admin.py b/products/replay_vision/backend/admin.py
index bf237068907f..06abb2caf5c2 100644
--- a/products/replay_vision/backend/admin.py
+++ b/products/replay_vision/backend/admin.py
@@ -26,6 +26,8 @@ class ReplayScannerAdmin(admin.ModelAdmin):
 @admin.register(ReplayObservation)
 class ReplayObservationAdmin(admin.ModelAdmin):
     list_display = ("scanner", "session_id", "status", "triggered_by", "created_at", "completed_at")
+    # The product's largest table: without this the changelist runs a query per row for the FK columns.
+    list_select_related = ("scanner", "team")
     list_filter = ("status", "triggered_by")
     search_fields = ("session_id", "workflow_id")
     # Observations are workflow-created and immutable post-create except for status/error_reason.
@@ -80,6 +82,7 @@ class VisionActionAdmin(admin.ModelAdmin):
         "next_run_at",
         "created_at",
     )
+    list_select_related = ("team", "scanner")
     list_filter = ("trigger_type", "mode", "enabled")
     search_fields = ("name",)
     raw_id_fields = ("team", "scanner", "hog_flow", "created_by")
@@ -89,6 +92,7 @@ class VisionActionAdmin(admin.ModelAdmin):
 @admin.register(VisionActionRun)
 class VisionActionRunAdmin(admin.ModelAdmin):
     list_display = ("id", "vision_action", "team", "status", "observation_count", "scheduled_at", "created_at")
+    list_select_related = ("vision_action", "team")
     list_filter = ("status",)
     search_fields = ("idempotency_key", "temporal_workflow_id")
     raw_id_fields = ("vision_action", "team")
diff --git a/products/replay_vision/backend/api/errors.py b/products/replay_vision/backend/api/errors.py
new file mode 100644
index 000000000000..04ccb9990c45
--- /dev/null
+++ b/products/replay_vision/backend/api/errors.py
@@ -0,0 +1,7 @@
+from rest_framework import serializers
+
+
+class ReplayVisionErrorSerializer(serializers.Serializer):
+    """The shape every Replay Vision error response uses, so generated clients read one key."""
+
+    detail = serializers.CharField(help_text="Human-readable explanation of why the request was refused.")
diff --git a/products/replay_vision/backend/api/observations.py b/products/replay_vision/backend/api/observations.py
index 2c3222419b14..d947b87ad762 100644
--- a/products/replay_vision/backend/api/observations.py
+++ b/products/replay_vision/backend/api/observations.py
@@ -5,7 +5,7 @@
 from zoneinfo import ZoneInfo
 
 from django.conf import settings
-from django.db import transaction
+from django.db import IntegrityError, transaction
 from django.db.models import Case, CharField, FloatField, Func, IntegerField, Q, QuerySet, Value, When
 from django.db.models.fields.json import KeyTextTransform, KeyTransform
 from django.db.models.functions import Cast
@@ -14,7 +14,13 @@
 import structlog
 import django_filters
 from django_filters.rest_framework import DjangoFilterBackend
-from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field, extend_schema_view
+from drf_spectacular.utils import (
+    OpenApiParameter,
+    OpenApiResponse,
+    extend_schema,
+    extend_schema_field,
+    extend_schema_view,
+)
 from pydantic import ValidationError as PydanticValidationError
 from rest_framework import mixins, serializers, status, viewsets
 from rest_framework.decorators import action
@@ -30,6 +36,7 @@
 from posthog.renderers import ServerSentEventRenderer
 from posthog.utils import relative_date_parse
 
+from products.replay_vision.backend.api.errors import ReplayVisionErrorSerializer
 from products.replay_vision.backend.api.filters import MultiChoiceFilter, OrderByFilter, ordering_enum
 from products.replay_vision.backend.api.observation_progress import stream_observation_progress
 from products.replay_vision.backend.api.observation_stats import compute_observation_stats
@@ -37,9 +44,11 @@
     WorkflowStartOutcome,
     check_observation_quota,
     check_team_in_flight_capacity,
+    claim_apply_scanner_slot,
     start_apply_scanner_workflow,
 )
 from products.replay_vision.backend.billing import observation_credits_for_model
+from products.replay_vision.backend.enqueue_claims import release_enqueue_claim
 from products.replay_vision.backend.error_kinds import ERROR_REASON_HELP_TEXT
 from products.replay_vision.backend.feature_flag import ReplayVisionEnabledPermission
 from products.replay_vision.backend.models.replay_observation import (
@@ -864,7 +873,16 @@ def create_task(self, request: Request, **kwargs: Any) -> Response:
             locked.save(update_fields=["created_task_id"])
         return Response({"task_id": task_id}, status=status.HTTP_201_CREATED)
 
-    @extend_schema(request=None, responses={202: RetryResponseSerializer})
+    @extend_schema(
+        request=None,
+        responses={
+            202: RetryResponseSerializer,
+            409: OpenApiResponse(
+                response=ReplayVisionErrorSerializer, description="The previous run is still finishing."
+            ),
+            503: OpenApiResponse(response=ReplayVisionErrorSerializer, description="The retry couldn't be started."),
+        },
+    )
     @action(detail=True, methods=["post"], required_scopes=["replay_scanner:write", "session_recording:read"])
     def retry(self, request: Request, **kwargs: Any) -> Response:
         """Delete a failed observation and re-run its scanner on the same recording. Returns 202 with the workflow handle."""
@@ -873,32 +891,68 @@ def retry(self, request: Request, **kwargs: Any) -> Response:
         scanner = getattr(self, "_scanner_for_url_cache", None) or observation.scanner
         # Retry writes to the scanner; the session route's get_object only object-checks the observation row.
         self.check_object_permissions(self.request, scanner)
+        session_id = observation.session_id
+        original_pk = observation.pk
+        original_created_at = observation.created_at
         if observation.status != ObservationStatus.FAILED:
             raise ValidationError("Only failed observations can be retried.")
+        # Advisory, and deliberately outside the lock below: the atomic claim is the authoritative gate,
+        # and these two read enough to be worth keeping off a held row lock.
         check_observation_quota(self.team.organization_id, observation_credits_for_model(scanner.model))
         check_team_in_flight_capacity(self.team.id)
-        session_id = observation.session_id
-        # Free the UNIQUE(scanner, session_id) slot; the usage ledger is immutable, so the failed attempt stays counted.
-        original_pk = observation.pk
-        original_created_at = observation.created_at
-        observation.delete()
+        # Locked so two concurrent retries can't both pass the status check and both delete the row.
+        with transaction.atomic():
+            locked = ReplayObservation.objects.select_for_update().get(pk=original_pk, team_id=self.team_id)
+            if locked.status != ObservationStatus.FAILED:
+                raise ValidationError("Only failed observations can be retried.")
+            # Captured before the delete cascades it away: a run that never starts has to put the team's
+            # rating back with the row, not just the row.
+            original_label = ReplayObservationLabel.objects.filter(
+                observation_id=original_pk, team_id=locked.team_id
+            ).first()
+            label_created_at = original_label.created_at if original_label else None
+            label_updated_at = original_label.updated_at if original_label else None
+            # Claimed before the delete so a capped retry never touches the row, and so never cascades
+            # away the observation's shared label for a request that changes nothing.
+            workflow_id, claimed = claim_apply_scanner_slot(scanner, session_id)
+            if not claimed:
+                raise Throttled(detail="This team is at its in-flight observation limit. Try again in a few minutes.")
+            try:
+                # Free the UNIQUE(scanner, session_id) slot; the ledger is immutable, so the failed attempt
+                # stays counted.
+                locked.delete()
+            except Exception:
+                release_enqueue_claim(
+                    team_id=scanner.team_id, scanner_id=scanner.id, workflow_id=workflow_id, immediately=True
+                )
+                raise
         workflow_id, outcome = start_apply_scanner_workflow(
             scanner,
             session_id,
             triggered_by_user_id=cast(User, request.user).id,
             trigger=ObservationTrigger.RETRY,
+            slot_already_claimed=True,
         )
         if outcome is not WorkflowStartOutcome.STARTED:
-            # The replacement run never started, so restore the failed row (its shared label, if any, is lost
-            # to the cascade) instead of leaving the recording looking unscanned.
-            observation.pk = original_pk
-            observation.save(force_insert=True)
-            ReplayObservation.objects.filter(pk=original_pk, team_id=observation.team_id).update(
-                created_at=original_created_at
-            )
-        if outcome is WorkflowStartOutcome.CAPPED:
-            # The pre-check above passed on a snapshot; the atomic claim is the authoritative gate.
-            raise Throttled(detail="This team is at its in-flight observation limit. Try again in a few minutes.")
+            # The replacement run never started, so restore the failed row and its rating instead of leaving
+            # the recording looking unscanned and the team's feedback gone.
+            try:
+                with transaction.atomic():
+                    observation.pk = original_pk
+                    observation.save(force_insert=True)
+                    ReplayObservation.objects.filter(pk=original_pk, team_id=observation.team_id).update(
+                        created_at=original_created_at
+                    )
+                    if original_label is not None:
+                        original_label.save(force_insert=True)
+                        # auto_now_add/auto_now stamp the insert with now; the rating happened earlier.
+                        ReplayObservationLabel.objects.filter(pk=original_label.pk, team_id=observation.team_id).update(
+                            created_at=label_created_at, updated_at=label_updated_at
+                        )
+            except IntegrityError:
+                # A run we couldn't start is already persisting its own row for this (scanner, session);
+                # the recording isn't stranded, so report it as still finishing rather than 500ing.
+                outcome = WorkflowStartOutcome.ALREADY_RUNNING
         if outcome is WorkflowStartOutcome.ALREADY_RUNNING:
             # The prior run is still closing, so its deterministic id blocks the restart and no new row will appear.
             return Response(
@@ -948,16 +1002,23 @@ def label(self, request: Request, **kwargs: Any) -> Response:
             return Response(status=204)
         input_serializer = ReplayObservationLabelSerializer(data=request.data)
         input_serializer.is_valid(raise_exception=True)
-        # team_id in the lookup keeps the query team-scoped.
-        label, _ = ReplayObservationLabel.objects.update_or_create(
-            observation=observation,
-            team_id=observation.team_id,
-            defaults={
-                "is_correct": input_serializer.validated_data["is_correct"],
-                "feedback": input_serializer.validated_data.get("feedback", ""),
-                "created_by": user,
-            },
-        )
+        # Lock the parent row so two concurrent first-time ratings serialize: unlocked, both see no label,
+        # both insert, and the loser hits the OneToOne constraint as a 500.
+        with transaction.atomic():
+            # `only("pk")`: the lock is the point, and the full row drags its JSONB columns along.
+            ReplayObservation.objects.select_for_update().only("pk").filter(
+                pk=observation.pk, team_id=observation.team_id
+            ).first()
+            # team_id in the lookup keeps the query team-scoped.
+            label, _ = ReplayObservationLabel.objects.update_or_create(
+                observation=observation,
+                team_id=observation.team_id,
+                defaults={
+                    "is_correct": input_serializer.validated_data["is_correct"],
+                    "feedback": input_serializer.validated_data.get("feedback", ""),
+                    "created_by": user,
+                },
+            )
         # The core calibration signal: thumbs up/down on whether the scanner got the session right.
         report_user_action(
             user,
diff --git a/products/replay_vision/backend/api/prompt_suggestions.py b/products/replay_vision/backend/api/prompt_suggestions.py
index a7e28e9f11f6..f05f4983153a 100644
--- a/products/replay_vision/backend/api/prompt_suggestions.py
+++ b/products/replay_vision/backend/api/prompt_suggestions.py
@@ -401,8 +401,6 @@ def evaluate(self, request: Request, **kwargs: Any) -> Response:
         input_serializer.is_valid(raise_exception=True)
         session_limit = input_serializer.validated_data["session_limit"]
         edited_config = input_serializer.validated_data.get("config")
-        if suggestion.status != SuggestionStatus.PENDING:
-            raise ValidationError("Only the current pending suggestion can be tested.")
         if not evaluation_supported(scanner):
             raise ValidationError("Testing isn't available for this scanner type.")
         # A malformed edited config must be rejected before it charges credits on runs that can't succeed,
@@ -414,29 +412,38 @@ def evaluate(self, request: Request, **kwargs: Any) -> Response:
         rated_count = self._rated_count(scanner)
         if rated_count == 0:
             raise ValidationError("Rate some results first. They are what the suggestion is tested against.")
-        # A test already in flight keeps reporting its state even if quota ran out meanwhile.
-        if evaluation_in_flight(suggestion.evaluation):
-            return Response(ReplayScannerPromptSuggestionSerializer(suggestion).data)
-        # Each re-run session charges credits like a normal observation, so refuse a test that would
-        # overspend the month. An uncapped org (no credit limit) never trips this.
-        planned = min(session_limit, rated_count)
-        planned_credits = planned * observation_credits_for_model(scanner.model)
-        quota = compute_quota_snapshot(organization_id=self.team.organization_id)
-        if quota.would_exceed(planned_credits):
-            raise QuotaLimitExceeded(
-                detail=(
-                    f"This test would use {planned_credits:,} credits but only {quota.remaining or 0:,} of the "
-                    f"monthly Replay Vision credit limit of {quota.credit_limit or 0:,} remain. Lower the test "
-                    f"session count or wait for the reset on "
-                    f"{quota.period_end.strftime('%b')} {quota.period_end.day}."
-                )
+        # Guards must run on a locked row, like apply and dismiss: on unlocked reads two concurrent tests
+        # both see "not in flight", and the second stub save moves `started_at`, which re-keys the usage
+        # receipts of the first run's still-settling sessions and charges them twice.
+        with transaction.atomic():
+            suggestion = ReplayScannerPromptSuggestion.objects.select_for_update().get(
+                team_id=self.team_id, id=suggestion.id
             )
-
-        # Stamp running first so the UI never sees a gap and the planned spend counts against quota
-        # right away. The select activity replaces this stub with the real total and fingerprint.
-        previous_evaluation = suggestion.evaluation
-        suggestion.evaluation = build_running_evaluation(total=planned, labels_fingerprint="")
-        suggestion.save(update_fields=["evaluation"])
+            if suggestion.status != SuggestionStatus.PENDING:
+                raise ValidationError("Only the current pending suggestion can be tested.")
+            # A test already in flight keeps reporting its state even if quota ran out meanwhile.
+            if evaluation_in_flight(suggestion.evaluation):
+                return Response(ReplayScannerPromptSuggestionSerializer(suggestion).data)
+            # Each re-run session charges credits like a normal observation, so refuse a test that would
+            # overspend the month. An uncapped org (no credit limit) never trips this.
+            planned = min(session_limit, rated_count)
+            planned_credits = planned * observation_credits_for_model(scanner.model)
+            quota = compute_quota_snapshot(organization_id=self.team.organization_id)
+            if quota.would_exceed(planned_credits):
+                raise QuotaLimitExceeded(
+                    detail=(
+                        f"This test would use {planned_credits:,} credits but only {quota.remaining or 0:,} of the "
+                        f"monthly Replay Vision credit limit of {quota.credit_limit or 0:,} remain. Lower the test "
+                        f"session count or wait for the reset on "
+                        f"{quota.period_end.strftime('%b')} {quota.period_end.day}."
+                    )
+                )
+            # Stamp running first so the UI never sees a gap and the planned spend counts against quota
+            # right away. The select activity replaces this stub with the real total and fingerprint.
+            previous_evaluation = suggestion.evaluation
+            suggestion.evaluation = build_running_evaluation(total=planned, labels_fingerprint="", model=scanner.model)
+            suggestion.save(update_fields=["evaluation"])
+        started_at = str(suggestion.evaluation.get("started_at") or "")
         try:
             client = sync_connect()
             async_to_sync(client.start_workflow)(  # type: ignore[misc]
@@ -446,6 +453,7 @@ def evaluate(self, request: Request, **kwargs: Any) -> Response:
                     team_id=scanner.team_id,
                     session_limit=session_limit,
                     config_override=edited_config,
+                    started_at=started_at,
                 ),
                 id=build_evaluate_prompt_suggestion_workflow_id(suggestion.id),
                 task_queue=settings.REPLAY_VISION_TASK_QUEUE,
diff --git a/products/replay_vision/backend/api/scanners.py b/products/replay_vision/backend/api/scanners.py
index 32f502b28cc3..4f7d9eb3c768 100644
--- a/products/replay_vision/backend/api/scanners.py
+++ b/products/replay_vision/backend/api/scanners.py
@@ -8,7 +8,13 @@
 import structlog
 import django_filters
 from django_filters.rest_framework import DjangoFilterBackend
-from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field, extend_schema_view
+from drf_spectacular.utils import (
+    OpenApiParameter,
+    OpenApiResponse,
+    extend_schema,
+    extend_schema_field,
+    extend_schema_view,
+)
 from pydantic import ValidationError as PydanticValidationError
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
@@ -26,6 +32,7 @@
 from posthog.rbac.access_control_api_mixin import AccessControlViewSetMixin
 from posthog.rbac.user_access_control import UserAccessControlSerializerMixin
 
+from products.replay_vision.backend.api.errors import ReplayVisionErrorSerializer
 from products.replay_vision.backend.api.filters import (
     MultiChoiceFilter,
     OrderByFilter,
@@ -1128,7 +1135,12 @@ def stats(self, request: Request, **kwargs: Any) -> Response:
 
     @extend_schema(
         request=ObserveRequestSerializer,
-        responses={202: ObserveResponseSerializer},
+        responses={
+            202: ObserveResponseSerializer,
+            503: OpenApiResponse(
+                response=ReplayVisionErrorSerializer, description="The observation workflow couldn't be started."
+            ),
+        },
     )
     @action(
         detail=True,
@@ -1163,7 +1175,8 @@ def observe(self, request: Request, **kwargs: Any) -> Response:
             raise Throttled(detail="This team is at its in-flight observation limit. Try again in a few minutes.")
         if outcome is WorkflowStartOutcome.FAILED:
             return Response(
-                {"error": "Failed to start observation workflow"},
+                # `detail` (not `error`) so ApiError carries the message into the frontend toast.
+                {"detail": "Failed to start the observation. Try again in a moment."},
                 status=status.HTTP_503_SERVICE_UNAVAILABLE,
             )
 
@@ -1292,8 +1305,9 @@ def _bulk_observe_headroom(self, scanner: ReplayScanner) -> tuple[int, str, int,
         )
         snapshot = compute_quota_snapshot(self.team.organization_id)
         cost = observation_credits_for_model(scanner.model)
-        # Uncapped org (remaining None) → quota never binds; otherwise how many of THIS model's cost fit.
-        quota_limit = in_flight_limit if snapshot.remaining is None else (snapshot.remaining // cost if cost else 0)
+        # Uncapped org, or a free model that spends nothing: quota can't bind. Otherwise, how many of
+        # THIS model's cost fit.
+        quota_limit = in_flight_limit if snapshot.remaining is None or cost <= 0 else snapshot.remaining // cost
         # Report quota as the reason only when it's the strictly tighter limit.
         if quota_limit < in_flight_limit:
             return quota_limit, "skipped_quota", team_in_flight, scanner_in_flight
@@ -1450,7 +1464,12 @@ def estimate(self, request: Request, **kwargs: Any) -> Response:
 
     @extend_schema(
         request=SuggestTagsRequestSerializer,
-        responses={200: SuggestTagsResponseSerializer},
+        responses={
+            200: SuggestTagsResponseSerializer,
+            503: OpenApiResponse(
+                response=ReplayVisionErrorSerializer, description="Tag suggestions couldn't be generated."
+            ),
+        },
     )
     @action(
         detail=False,
@@ -1489,7 +1508,7 @@ def suggest_tags(self, request: Request, **kwargs: Any) -> Response:
             )
         except SuggestionError:
             return Response(
-                {"error": "Couldn't generate tag suggestions right now. Please try again."},
+                {"detail": "Couldn't generate tag suggestions right now. Try again in a moment."},
                 status=status.HTTP_503_SERVICE_UNAVAILABLE,
             )
 
diff --git a/products/replay_vision/backend/api/trigger.py b/products/replay_vision/backend/api/trigger.py
index aba89499e056..c55d4792fcaf 100644
--- a/products/replay_vision/backend/api/trigger.py
+++ b/products/replay_vision/backend/api/trigger.py
@@ -86,17 +86,16 @@ def _admission_still_within_caps(scanner: ReplayScanner) -> bool:
     return scanner_rows + pending_enqueue_claims_for_scanner(scanner.id) <= MAX_IN_FLIGHT_APPLIES_PER_SCANNER
 
 
-def start_apply_scanner_workflow(
+def claim_apply_scanner_slot(
     scanner: ReplayScanner,
     session_id: str,
     *,
-    triggered_by_user_id: int,
-    trigger: ObservationTrigger,
     team_in_flight_rows: int | None = None,
     scanner_in_flight_rows: int | None = None,
-) -> tuple[str, WorkflowStartOutcome]:
-    """Start the deterministic apply-scanner workflow for one (scanner, session); never raises.
-    An atomic enqueue-slot claim guards the in-flight caps; pass row counts to save two queries."""
+) -> tuple[str, bool]:
+    """Claim the enqueue slot for one (scanner, session) ahead of the workflow start; on success the
+    caller owns the claim and must either pass `slot_already_claimed=True` to
+    `start_apply_scanner_workflow` or release it."""
     workflow_id = build_apply_scanner_workflow_id(scanner.id, session_id)
     if team_in_flight_rows is None:
         team_in_flight_rows = ReplayObservation.in_flight_for_team(scanner.team_id).count()
@@ -111,10 +110,36 @@ def start_apply_scanner_workflow(
         team_in_flight_rows=team_in_flight_rows,
         scanner_in_flight_rows=scanner_in_flight_rows,
     ):
-        return workflow_id, WorkflowStartOutcome.CAPPED
+        return workflow_id, False
     if not _admission_still_within_caps(scanner):
         release_enqueue_claim(team_id=scanner.team_id, scanner_id=scanner.id, workflow_id=workflow_id)
-        return workflow_id, WorkflowStartOutcome.CAPPED
+        return workflow_id, False
+    return workflow_id, True
+
+
+def start_apply_scanner_workflow(
+    scanner: ReplayScanner,
+    session_id: str,
+    *,
+    triggered_by_user_id: int,
+    trigger: ObservationTrigger,
+    team_in_flight_rows: int | None = None,
+    scanner_in_flight_rows: int | None = None,
+    slot_already_claimed: bool = False,
+) -> tuple[str, WorkflowStartOutcome]:
+    """Start the deterministic apply-scanner workflow for one (scanner, session); never raises.
+    An atomic enqueue-slot claim guards the in-flight caps; pass row counts to save two queries."""
+    if slot_already_claimed:
+        workflow_id = build_apply_scanner_workflow_id(scanner.id, session_id)
+    else:
+        workflow_id, claimed = claim_apply_scanner_slot(
+            scanner,
+            session_id,
+            team_in_flight_rows=team_in_flight_rows,
+            scanner_in_flight_rows=scanner_in_flight_rows,
+        )
+        if not claimed:
+            return workflow_id, WorkflowStartOutcome.CAPPED
     try:
         client = sync_connect()
         async_to_sync(client.start_workflow)(  # type: ignore[misc]
diff --git a/products/replay_vision/backend/api/vision_actions.py b/products/replay_vision/backend/api/vision_actions.py
index 98a4a04427fb..62505b8bfdc8 100644
--- a/products/replay_vision/backend/api/vision_actions.py
+++ b/products/replay_vision/backend/api/vision_actions.py
@@ -10,7 +10,13 @@
 
 import structlog
 from drf_spectacular.types import OpenApiTypes
-from drf_spectacular.utils import OpenApiParameter, extend_schema, extend_schema_field, extend_schema_view
+from drf_spectacular.utils import (
+    OpenApiParameter,
+    OpenApiResponse,
+    extend_schema,
+    extend_schema_field,
+    extend_schema_view,
+)
 from rest_framework import mixins, serializers, status, viewsets
 from rest_framework.decorators import action
 from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError
@@ -25,6 +31,7 @@
 from posthog.models.user import User
 
 from products.replay_vision.backend.api.delivery import archive_delivery, provision_delivery
+from products.replay_vision.backend.api.errors import ReplayVisionErrorSerializer
 from products.replay_vision.backend.api.trigger import WorkflowStartOutcome, start_process_vision_action_workflow
 from products.replay_vision.backend.feature_flag import (
     ReplayVisionActionsEnabledPermission,
@@ -726,7 +733,15 @@ def perform_destroy(self, instance: VisionAction) -> None:
         archive_delivery(instance, team=self.team)
         super().perform_destroy(instance)
 
-    @extend_schema(request=None, responses={202: RunActionResponseSerializer})
+    @extend_schema(
+        request=None,
+        responses={
+            202: RunActionResponseSerializer,
+            503: OpenApiResponse(
+                response=ReplayVisionErrorSerializer, description="The summary run couldn't be started."
+            ),
+        },
+    )
     @action(
         detail=True,
         methods=["post"],
diff --git a/products/replay_vision/backend/observation_formatting.py b/products/replay_vision/backend/observation_formatting.py
index 63146bc56fb2..aba30df6836a 100644
--- a/products/replay_vision/backend/observation_formatting.py
+++ b/products/replay_vision/backend/observation_formatting.py
@@ -38,7 +38,8 @@ def format_line(obs: "ReplayObservation", output: dict[str, Any], *, show_scanne
     descriptor = describe_output(output)
     explanation = output.get("reasoning") or output.get("summary")
     if not isinstance(explanation, str) or not explanation.strip():
-        explanation = output.get("intent") or output.get("outcome") or ""
+        fallback = output.get("intent") or output.get("outcome")
+        explanation = fallback if isinstance(fallback, str) else ""
     clean = re.sub(r"\s+", " ", EVENT_ID_CITATION_RE.sub("", explanation)).strip()[:SEARCH_SNIPPET_LIMIT]
 
     prefix = f"{obs.created_at:%Y-%m-%d}"
diff --git a/products/replay_vision/backend/prompt_evaluation.py b/products/replay_vision/backend/prompt_evaluation.py
index 630b4c88fa1e..b4e942a7401f 100644
--- a/products/replay_vision/backend/prompt_evaluation.py
+++ b/products/replay_vision/backend/prompt_evaluation.py
@@ -12,6 +12,7 @@
 from products.replay_vision.backend.models.replay_observation import ObservationStatus, ReplayObservation
 from products.replay_vision.backend.models.replay_scanner import ReplayScanner, ScannerType
 from products.replay_vision.backend.models.replay_scanner_prompt_suggestion import ReplayScannerPromptSuggestion
+from products.replay_vision.backend.tags import slugify_tag
 
 # Sized for a full run at the session cap (100 sessions, 4 concurrent, a few minutes each).
 # Lives here rather than temporal/constants so quota-path imports don't drag in the temporal package.
@@ -78,7 +79,10 @@ def primary_outcome(model_output: dict[str, Any] | None) -> str | None:
     verdict = output.get("verdict")
     if isinstance(verdict, str) and verdict:
         return f"Verdict: {verdict.strip().lower()}"
-    tags = sorted(t.strip().lower() for t in (output.get("tags") or []) if isinstance(t, str) and t.strip())
+    # Freeform tags are part of a classifier's output, so a rewrite that only changes them must not read as
+    # "no change". Matches `describe_output`, which merges both lists.
+    raw_tags = [*(output.get("tags") or []), *(output.get("tags_freeform") or [])]
+    tags = sorted({slug for t in raw_tags if isinstance(t, str) and (slug := slugify_tag(t))})
     if tags:
         return f"Tags: {', '.join(tags)}"
     # Preview types have no discrete outcome, so show the raw output the reviewer compares by eye.
@@ -126,21 +130,31 @@ def in_flight_evaluation_credits(organization_id: uuid.UUID) -> int:
         team__organization_id=organization_id, evaluation__status="running"
     ).values_list("evaluation", "scanner__model")
     total = 0
-    for evaluation, model in rows:
+    for evaluation, scanner_model in rows:
         if not isinstance(evaluation, dict) or not evaluation_in_flight(evaluation):
             continue
         unsettled = max(0, int(evaluation.get("total") or 0) - len(evaluation.get("results") or []))
+        # Receipts bill the model frozen at workflow start, so the reservation prices from the same frozen
+        # value. Stubs written before the field existed fall back to the scanner's current model.
+        model = evaluation.get("model") or scanner_model
         total += unsettled * observation_credits_for_model(model or "")
     return total
 
 
-def build_running_evaluation(total: int, labels_fingerprint: str) -> dict[str, Any]:
+def build_running_evaluation(
+    total: int,
+    labels_fingerprint: str,
+    model: str | None = None,
+    started_at: str | None = None,
+) -> dict[str, Any]:
     return {
         "status": "running",
-        "started_at": timezone.now().isoformat(),
+        # Usage receipt ids are keyed on this, so a restamp mid-run must reuse the original value.
+        "started_at": started_at or timezone.now().isoformat(),
         "finished_at": None,
         "total": total,
         "labels_fingerprint": labels_fingerprint,
+        "model": model,
         "results": [],
         "summary": None,
     }
diff --git a/products/replay_vision/backend/prompt_suggestions.py b/products/replay_vision/backend/prompt_suggestions.py
index 9d9854f232e1..9ea11cc9fdd3 100644
--- a/products/replay_vision/backend/prompt_suggestions.py
+++ b/products/replay_vision/backend/prompt_suggestions.py
@@ -54,6 +54,10 @@
 _MAX_REASONING_CHARS = 280
 _MAX_DISMISSED_EXAMPLES = 3
 _MAX_DISMISSED_PROMPT_CHARS = 600
+# `feedback` is an unbounded TextField, so one pasted log dump would otherwise balloon every future
+# briefing for that scanner (and the per-session tool payload).
+_MAX_BRIEFING_FEEDBACK_CHARS = 600
+_MAX_TOOL_FEEDBACK_CHARS = 2000
 _MAX_TOOL_ROUNDS = 6
 _MAX_SUMMARIES_PER_RUN = 2
 _MAX_TOOL_REASONING_CHARS = 4000
@@ -122,7 +126,17 @@ def _describe_reasoning(observation: ReplayObservation) -> str:
     reasoning = output.get("reasoning")
     if not isinstance(reasoning, str) or not reasoning:
         return ""
-    return reasoning[:_MAX_REASONING_CHARS] + ("…" if len(reasoning) > _MAX_REASONING_CHARS else "")
+    return _clip(reasoning, _MAX_REASONING_CHARS)
+
+
+def _defuse_fence(text: str) -> str:
+    """A rewrite containing \"\"\" would close its own fence early and have the remainder read as briefing
+    instructions."""
+    return text.replace('"""', "'''")
+
+
+def _clip(text: str, limit: int) -> str:
+    return text[:limit] + "…" if len(text) > limit else text
 
 
 def _label(observation: ReplayObservation) -> ReplayObservationLabel:
@@ -134,7 +148,8 @@ def _example_line(observation: ReplayObservation) -> str:
     label = _label(observation)
     parts = [f"- Session {observation.session_id}. Scanner output: {_describe_outcome(observation)}"]
     if label.feedback:
-        parts.append(f"{'What it should be' if not label.is_correct else 'Note'}: {label.feedback}")
+        feedback = _clip(label.feedback, _MAX_BRIEFING_FEEDBACK_CHARS)
+        parts.append(f"{'What it should be' if not label.is_correct else 'Note'}: {feedback}")
     reasoning = _describe_reasoning(observation)
     if reasoning:
         parts.append(f"Its reasoning: {reasoning}")
@@ -188,10 +203,8 @@ def _dismissed_lines(scanner: ReplayScanner) -> list[str]:
         "Previously rejected rewrites (the team dismissed these; do not propose them again or close variations):",
     ]
     for suggestion in dismissed:
-        prompt = suggestion.suggested_prompt
-        if len(prompt) > _MAX_DISMISSED_PROMPT_CHARS:
-            prompt = prompt[:_MAX_DISMISSED_PROMPT_CHARS] + "…"
-        lines.append(f'- """{prompt}"""')
+        prompt = _clip(suggestion.suggested_prompt, _MAX_DISMISSED_PROMPT_CHARS)
+        lines.append(f'- """{_defuse_fence(prompt)}"""')
     return lines
 
 
@@ -206,7 +219,7 @@ def _build_user_content(
         "",
         "Current prompt:",
         '"""',
-        str(base_config.get("prompt", "")),
+        _defuse_fence(str(base_config.get("prompt", ""))),
         '"""',
     ]
     if wrong:
@@ -369,7 +382,7 @@ def _tool_get_rated_observation(state: _AgentToolState, session_id: str) -> dict
         "output": _describe_outcome(observation),
         "reasoning": reasoning[:_MAX_TOOL_REASONING_CHARS],
         "rating": "thumbs_up" if label.is_correct else "thumbs_down",
-        "feedback": label.feedback,
+        "feedback": _clip(label.feedback, _MAX_TOOL_FEEDBACK_CHARS),
         "prompt_version": snapshot.get("scanner_version"),
     }
 
diff --git a/products/replay_vision/backend/proposers/classifier.py b/products/replay_vision/backend/proposers/classifier.py
index 9c562946f73b..07bdc2d6c654 100644
--- a/products/replay_vision/backend/proposers/classifier.py
+++ b/products/replay_vision/backend/proposers/classifier.py
@@ -1,3 +1,5 @@
+from collections import deque
+from collections.abc import Iterator
 from typing import TYPE_CHECKING, Any
 
 from products.replay_vision.backend import tag_suggestions
@@ -58,7 +60,7 @@ def grounding(self, scanner: "ReplayScanner") -> str:
     def to_config_patch(self, llm_output: dict[str, Any], base_config: dict[str, Any]) -> dict[str, Any]:
         config = dict(base_config)
         config["prompt"] = str(llm_output["suggested_prompt"]).strip()
-        config["tags"] = _apply_tag_ops(list(base_config.get("tags", [])), llm_output.get("tag_ops", []))
+        config["tags"] = _apply_tag_ops(list(base_config.get("tags", [])), _valid_tag_ops(llm_output.get("tag_ops")))
         return config
 
     def to_changes(
@@ -66,29 +68,13 @@ def to_changes(
     ) -> list[ConfigChange]:
         rationale = str(llm_output.get("rationale", "")).strip()
         changes = prompt_change(base_config, suggested_config, rationale)
-        # Emit a change only for an op that actually alters the vocabulary, so a no-op op does not mark an
-        # unchanged config as pending.
         working = list(base_config.get("tags", []))
-        for op in llm_output.get("tag_ops", []):
-            kind, tag, to = op.get("op"), op.get("tag"), op.get("to")
-            if not kind or not tag:
-                continue
-            if kind == "add" and tag not in working:
-                working.append(tag)
-                before, after = None, tag
-            elif kind == "remove" and tag in working:
-                working.remove(tag)
-                before, after = tag, None
-            elif kind == "rename" and tag in working and to:
-                _rename_tag(working, tag, to)
-                before, after = tag, to
-            else:
-                continue
+        for op, before, after in _tag_transitions(working, _valid_tag_ops(llm_output.get("tag_ops"))):
             changes.append(
                 ConfigChange(
                     field="tags",
                     kind="tags",
-                    op=kind,
+                    op=op["op"],
                     before=before,
                     after=after,
                     rationale=str(op.get("rationale", "")),
@@ -97,30 +83,60 @@ def to_changes(
         return changes
 
 
-def _apply_tag_ops(tags: list[str], ops: list[dict[str, Any]]) -> list[str]:
-    result = list(tags)
-    for op in ops:
-        kind, tag = op.get("op"), op.get("tag")
-        # A malformed op (schema not honored) is skipped rather than raising, so one bad op can't turn a
-        # whole generation into a 500 instead of a usable suggestion.
-        if not kind or not tag:
+def _valid_tag_ops(raw: Any) -> list[dict[str, Any]]:
+    """The schema guides the model, it doesn't bind it, and this runs outside the generation fallbacks:
+    one malformed op used to lose the whole suggestion, rewritten prompt included."""
+    if not isinstance(raw, list):
+        return []
+    ops: list[dict[str, Any]] = []
+    for op in raw:
+        if not isinstance(op, dict):
             continue
-        if kind == "add" and tag not in result:
-            result.append(tag)
-        elif kind == "remove" and tag in result:
-            result.remove(tag)
-        elif kind == "rename" and tag in result and op.get("to"):
-            _rename_tag(result, tag, op["to"])
-    return result
+        kind, tag, to = op.get("op"), op.get("tag"), op.get("to")
+        if kind not in ("add", "remove", "rename"):
+            continue
+        if not isinstance(tag, str) or not tag.strip():
+            continue
+        if kind == "rename" and (not isinstance(to, str) or not to.strip()):
+            continue
+        ops.append(op)
+    return ops
 
 
-def _rename_tag(tags: list[str], tag: str, to: str) -> None:
-    """Rename in place, but merge into the destination when another tag already shares its slug rather than
-    creating a duplicate. Tag uniqueness is slug-normalized (see api.scanners), so a plain string check would
-    still let `Payment` and `payment` both land and make the suggestion fail to apply."""
-    index = tags.index(tag)
-    to_slug = slugify_tag(to)
-    if any(slugify_tag(other) == to_slug for i, other in enumerate(tags) if i != index):
-        tags.pop(index)
-    else:
-        tags[index] = to
+def _slug_taken(tags: list[str], candidate: str, *, skip_index: int | None = None) -> bool:
+    """Tag uniqueness is slug-normalized (see api.scanners), so a plain string check would let `Payment`
+    and `payment` both land and make the suggestion impossible to apply."""
+    slug = slugify_tag(candidate)
+    return any(slugify_tag(other) == slug for i, other in enumerate(tags) if i != skip_index)
+
+
+def _tag_transitions(
+    tags: list[str], ops: list[dict[str, Any]]
+) -> Iterator[tuple[dict[str, Any], str | None, str | None]]:
+    """Apply each op to `tags` in place, yielding (op, before, after) only for ops that changed the
+    vocabulary — a no-op op must not mark an unchanged config as pending."""
+    for op in ops:
+        kind, tag = op["op"], op["tag"]
+        if kind == "add":
+            if not slugify_tag(tag) or _slug_taken(tags, tag):
+                continue
+            tags.append(tag)
+            yield op, None, tag
+        elif kind == "remove" and tag in tags:
+            tags.remove(tag)
+            yield op, tag, None
+        elif kind == "rename" and tag in tags:
+            index = tags.index(tag)
+            to = op["to"]
+            # Merge rather than duplicate when the destination slug is already present.
+            if _slug_taken(tags, to, skip_index=index):
+                tags.pop(index)
+            else:
+                tags[index] = to
+            yield op, tag, to
+
+
+def _apply_tag_ops(tags: list[str], ops: list[dict[str, Any]]) -> list[str]:
+    result = list(tags)
+    deque(_tag_transitions(result, ops), maxlen=0)  # Drain for the in-place edits; the yields are for to_changes.
+    return result
diff --git a/products/replay_vision/backend/proposers/scorer.py b/products/replay_vision/backend/proposers/scorer.py
index b3e1d964bc01..c87b4b9f3cc5 100644
--- a/products/replay_vision/backend/proposers/scorer.py
+++ b/products/replay_vision/backend/proposers/scorer.py
@@ -1,6 +1,7 @@
 from typing import TYPE_CHECKING, Any
 
 from products.replay_vision.backend.proposers.base import ConfigChange, prompt_change, set_change
+from products.replay_vision.backend.temporal.scanners.scorer import DEFAULT_SCORE_SCALE
 
 if TYPE_CHECKING:
     from products.replay_vision.backend.models.replay_scanner import ReplayScanner
@@ -45,10 +46,12 @@ def system_prompt(self) -> str:
         return _SYSTEM_PROMPT
 
     def grounding(self, scanner: "ReplayScanner") -> str:
-        # The schema requires echoing the scale, so the briefing must state the current one.
+        # The schema requires echoing the scale, so the briefing must state the current one. Staying silent
+        # when none is stored makes the model invent a range that `to_config_patch` then materializes as a
+        # deliberate change complete with a rationale.
         scale = (scanner.scanner_config or {}).get("scale") or {}
         if not scale:
-            return ""
+            return f"Current scale: none stored, so {DEFAULT_SCORE_SCALE.min} to {DEFAULT_SCORE_SCALE.max} applies. Echo it unless the rated sessions justify a different range."
         label = f" ({scale['label']})" if scale.get("label") else ""
         return f"Current scale: {scale.get('min')} to {scale.get('max')}{label}."
 
@@ -62,8 +65,8 @@ def to_config_patch(self, llm_output: dict[str, Any], base_config: dict[str, Any
         min_value = scale.get("min")
         max_value = scale.get("max")
         config["scale"] = {
-            "min": float(min_value if min_value is not None else base_scale.get("min", 0.0)),
-            "max": float(max_value if max_value is not None else base_scale.get("max", 0.0)),
+            "min": float(min_value if min_value is not None else base_scale.get("min", DEFAULT_SCORE_SCALE.min)),
+            "max": float(max_value if max_value is not None else base_scale.get("max", DEFAULT_SCORE_SCALE.max)),
             "label": scale.get("label", base_scale.get("label")),
         }
         return config
diff --git a/products/replay_vision/backend/quota.py b/products/replay_vision/backend/quota.py
index 9494c5cc45fe..fde5393e46dc 100644
--- a/products/replay_vision/backend/quota.py
+++ b/products/replay_vision/backend/quota.py
@@ -180,7 +180,8 @@ def compute_quota_snapshot(organization_id: UUID) -> QuotaSnapshot:
         observation_created_at__lt=period_end,
     ).aggregate(total=Coalesce(Sum("credits"), Value(0), output_field=IntegerField()))["total"]
     # In-flight rows aren't in the ledger yet (receipt is written on success), so reserve their credits live,
-    # priced from the frozen snapshot model exactly as the eventual receipt will be.
+    # priced from the frozen snapshot model exactly as the eventual receipt will be. One created just before
+    # a period rollover settles into the next window, so it is briefly counted in neither; accepted.
     in_flight_models = Counter(
         ReplayObservation.objects.filter(
             team__organization_id=organization_id,
diff --git a/products/replay_vision/backend/temporal/activities/call_scanner_provider.py b/products/replay_vision/backend/temporal/activities/call_scanner_provider.py
index db5f64bbce50..3e30dc035019 100644
--- a/products/replay_vision/backend/temporal/activities/call_scanner_provider.py
+++ b/products/replay_vision/backend/temporal/activities/call_scanner_provider.py
@@ -97,6 +97,7 @@ async def _call_scanner_provider(inputs: CallScannerProviderInputs) -> ScannerCa
         session_metadata=llm_inputs.metadata.as_prompt_dict(),
         navigation=[entry.model_dump() for entry in llm_inputs.navigation],
         navigation_dropped=llm_inputs.navigation_dropped,
+        events_truncated=llm_inputs.events_truncated,
     )
     video_part = types.Part(file_data=types.FileData(file_uri=inputs.file_uri, mime_type=inputs.mime_type))
 
diff --git a/products/replay_vision/backend/temporal/activities/create_observation.py b/products/replay_vision/backend/temporal/activities/create_observation.py
index b4c05e5f4fe1..88706db8477c 100644
--- a/products/replay_vision/backend/temporal/activities/create_observation.py
+++ b/products/replay_vision/backend/temporal/activities/create_observation.py
@@ -79,6 +79,8 @@ def _create_observation(inputs: CreateObservationInputs) -> CreateObservationOut
                 f"User {inputs.triggered_by_user_id} is not a member of scanner {inputs.scanner_id}'s organization"
             )
 
+    # Deliberately check-then-act: the snapshot doesn't count enqueue claims, so a concurrent burst can
+    # overshoot by at most the in-flight caps allow, which is accepted.
     if compute_quota_snapshot(scanner.team.organization_id).would_exceed(observation_credits_for_model(scanner.model)):
         record_quota_exhausted_skip(scanner.scanner_type)
         activity.logger.info(
diff --git a/products/replay_vision/backend/temporal/activities/emit_observation_signal.py b/products/replay_vision/backend/temporal/activities/emit_observation_signal.py
index 8bb84f6e6639..d4e67e6d35b3 100644
--- a/products/replay_vision/backend/temporal/activities/emit_observation_signal.py
+++ b/products/replay_vision/backend/temporal/activities/emit_observation_signal.py
@@ -83,6 +83,8 @@ def emit_observation_signal_activity(inputs: EmitObservationSignalInputs) -> int
                     source_type=VISION_SIGNALS_SOURCE_TYPE,
                     # Unique per finding so several issues from one observation don't collide on dedup.
                     source_id=f"observation:{observation.id}:{index}",
+                    # Same key on a retry, so a re-run after a lost attempt re-emits nothing.
+                    idempotency_key=f"observation:{observation.id}:{index}",
                     description=signal.description,
                     weight=SIGNAL_WEIGHT,
                     extra={
@@ -95,6 +97,8 @@ def emit_observation_signal_activity(inputs: EmitObservationSignalInputs) -> int
                     },
                 )
                 emitted += 1
+                # Findings go out one network call at a time, so beat between them.
+                activity.heartbeat({"emitted": emitted, "index": index})
             except Exception:
                 # One bad finding never blocks the rest; signals are advisory.
                 any_failed = True
diff --git a/products/replay_vision/backend/temporal/activities/evaluate_prompt_suggestion.py b/products/replay_vision/backend/temporal/activities/evaluate_prompt_suggestion.py
index 370161b8cffd..ff9789d47ab5 100644
--- a/products/replay_vision/backend/temporal/activities/evaluate_prompt_suggestion.py
+++ b/products/replay_vision/backend/temporal/activities/evaluate_prompt_suggestion.py
@@ -88,7 +88,10 @@ def select_evaluation_sessions_activity(inputs: SelectEvaluationSessionsInputs)
         scanner_config=rerun_config,
     )
     suggestion.evaluation = build_running_evaluation(
-        total=len(sessions), labels_fingerprint=labels_fingerprint(scanner)
+        total=len(sessions),
+        labels_fingerprint=labels_fingerprint(scanner),
+        model=scanner.model,
+        started_at=inputs.started_at,
     )
     suggestion.save(update_fields=["evaluation"])
     return SelectEvaluationSessionsOutput(sessions=sessions, snapshot=snapshot)
@@ -139,7 +142,9 @@ def record_evaluation_result_activity(inputs: RecordEvaluationResultInputs) -> N
                 observation_id=evaluation_usage_id(
                     suggestion.id,
                     inputs.session.session_id,
-                    str(suggestion.evaluation.get("started_at") or ""),
+                    # Frozen at request time so a concurrent re-test that restamps the row can't re-key
+                    # this run's receipts and charge its sessions twice.
+                    inputs.started_at or str(suggestion.evaluation.get("started_at") or ""),
                 ),
                 defaults={
                     "organization_id": suggestion.team.organization_id,
diff --git a/products/replay_vision/backend/temporal/activities/fetch_session_events.py b/products/replay_vision/backend/temporal/activities/fetch_session_events.py
index 491d79895fff..e04e0a824e6f 100644
--- a/products/replay_vision/backend/temporal/activities/fetch_session_events.py
+++ b/products/replay_vision/backend/temporal/activities/fetch_session_events.py
@@ -39,6 +39,9 @@
 # Events are no longer inlined in the prompt — they're loaded into the table the model queries on demand — so we
 # page through the whole session.
 _EVENTS_PER_PAGE = 2000
+# Eligibility caps active seconds, not event count, so an instrumentation loop or bot can still emit
+# millions of rows. Cap the total we hold in memory, gzip into Redis, and index for the events tool.
+_MAX_TOTAL_EVENT_ROWS = 50_000
 
 # Noisy SDK-internal events that add no signal for the LLM.
 _EVENTS_TO_IGNORE = ["$feature_flag_called"]
@@ -136,6 +139,7 @@ def _fetch_payload(team_id: int, session_id: str) -> ScannerLlmInputs | None:
 
     columns: list[str] | None = None
     all_rows: list[list[Any]] = []
+    events_truncated = False
     for page_number in itertools.count():
         page = events_obj.get_events(
             session_id=session_id,
@@ -151,6 +155,16 @@ def _fetch_payload(team_id: int, session_id: str) -> ScannerLlmInputs | None:
         if not page.rows:
             break
         all_rows.extend(list(row) for row in page.rows)
+        if len(all_rows) >= _MAX_TOTAL_EVENT_ROWS:
+            events_truncated = len(all_rows) > _MAX_TOTAL_EVENT_ROWS or page.has_more
+            del all_rows[_MAX_TOTAL_EVENT_ROWS:]
+            logger.warning(
+                "replay_vision.fetch.events_truncated",
+                session_id=session_id,
+                team_id=team_id,
+                cap=_MAX_TOTAL_EVENT_ROWS,
+            )
+            break
         if not page.has_more:
             break
 
@@ -170,6 +184,7 @@ def _fetch_payload(team_id: int, session_id: str) -> ScannerLlmInputs | None:
         event_timestamps=processed.event_timestamps,
         navigation=processed.navigation,
         navigation_dropped=processed.navigation_dropped,
+        events_truncated=events_truncated,
         distinct_id=metadata.get("distinct_id"),
         metadata=SessionMetadata(
             start_time=metadata["start_time"],
diff --git a/products/replay_vision/backend/temporal/activities/reap_orphaned_observations.py b/products/replay_vision/backend/temporal/activities/reap_orphaned_observations.py
index d827b44d12a5..ca408c80eda9 100644
--- a/products/replay_vision/backend/temporal/activities/reap_orphaned_observations.py
+++ b/products/replay_vision/backend/temporal/activities/reap_orphaned_observations.py
@@ -1,21 +1,35 @@
-"""Marks provably-orphaned pending/running observations as failed so they stop blocking re-scans and eating quota."""
+"""Marks provably-orphaned pending/running observations as failed so they stop blocking re-scans and eating quota.
+
+The same pass reaps prompt-suggestion evaluations stuck in `running`: the workflow swallows finalize
+failures, so a terminated or timed-out run leaves the row claiming to be in flight forever.
+"""
 
 from datetime import UTC, datetime
 from typing import Any, cast
 from uuid import UUID
 
+from django.db import transaction
+from django.utils import timezone
+
 import structlog
 from temporalio import activity
+from temporalio.client import Client
 
 from posthog.sync import database_sync_to_async
 from posthog.temporal.common.client import async_connect
 
 from products.replay_vision.backend.models.replay_observation import ObservationStatus, ReplayObservation
+from products.replay_vision.backend.models.replay_scanner_prompt_suggestion import ReplayScannerPromptSuggestion
+from products.replay_vision.backend.prompt_evaluation import (
+    EVALUATE_PROMPT_SUGGESTION_EXECUTION_TIMEOUT,
+    summarize_results,
+)
 from products.replay_vision.backend.temporal.activities.observation_state import mark_observation_terminal
 from products.replay_vision.backend.temporal.activities.reaping import classify_stale_rows
 from products.replay_vision.backend.temporal.constants import (
     OBSERVATION_ORPHAN_CUTOFF,
     REAP_ORPHANED_OBSERVATIONS_BATCH_SIZE,
+    build_evaluate_prompt_suggestion_workflow_id,
 )
 from products.replay_vision.backend.temporal.decorators import track_activity
 from products.replay_vision.backend.temporal.errors import FailureKind
@@ -25,6 +39,10 @@
 
 _LIVE_STATUSES = (ObservationStatus.PENDING, ObservationStatus.RUNNING)
 _ORPHANED_ERROR_REASON = f"{FailureKind.ORPHANED.value}:The analysis stopped without recording an outcome."
+# Evaluations whose stamp predates the workflow's own execution timeout can no longer be live.
+_EVALUATION_ORPHAN_CUTOFF = EVALUATE_PROMPT_SUGGESTION_EXECUTION_TIMEOUT * 2
+# One pass reaps at most this many; the rest drain on later reconciler ticks.
+_EVALUATION_BATCH_SIZE = 100
 
 
 def _list_stale_observations() -> list[dict[str, Any]]:
@@ -38,6 +56,30 @@ def _list_stale_observations() -> list[dict[str, Any]]:
     return cast(list[dict[str, Any]], list(rows))
 
 
+def _evaluation_stamp_is_stale(evaluation: dict[str, Any], cutoff: datetime) -> bool:
+    """A stamp that is unparseable or naive can never age out on its own, so it counts as stale."""
+    try:
+        started_at = datetime.fromisoformat(str(evaluation.get("started_at") or ""))
+    except ValueError:
+        return True
+    return started_at.tzinfo is None or started_at < cutoff
+
+
+def _list_stale_evaluations() -> list[dict[str, Any]]:
+    """Suggestions whose running evaluation predates the cutoff, shaped for `classify_stale_rows`."""
+    cutoff = timezone.now() - _EVALUATION_ORPHAN_CUTOFF
+    rows = (
+        ReplayScannerPromptSuggestion.objects.filter(evaluation__status="running")
+        .order_by("created_at")
+        .values("id", "evaluation")[:_EVALUATION_BATCH_SIZE]
+    )
+    return [
+        {"id": row["id"], "workflow_id": build_evaluate_prompt_suggestion_workflow_id(row["id"])}
+        for row in rows
+        if isinstance(row["evaluation"], dict) and _evaluation_stamp_is_stale(row["evaluation"], cutoff)
+    ]
+
+
 def _mark_orphaned(observation_id: UUID, scanner_type: str) -> bool:
     return mark_observation_terminal(
         observation_id=observation_id,
@@ -49,20 +91,31 @@ def _mark_orphaned(observation_id: UUID, scanner_type: str) -> bool:
     )
 
 
-@activity.defn
-@track_activity()
-async def reap_orphaned_observations_activity() -> int:
-    """Fail pending/running rows past the orphan cutoff whose workflow is no longer open; returns the count reaped.
+def _fail_evaluation(suggestion_id: UUID) -> bool:
+    """Settle a stuck evaluation as failed. Mirrors `finalize_evaluation_activity`'s write."""
+    with transaction.atomic():
+        suggestion = ReplayScannerPromptSuggestion.objects.select_for_update().filter(pk=suggestion_id).first()
+        if suggestion is None or not isinstance(suggestion.evaluation, dict):
+            return False
+        # A run that finalized between the listing and this write must not be clobbered.
+        if suggestion.evaluation.get("status") != "running":
+            return False
+        results = suggestion.evaluation.get("results", [])
+        suggestion.evaluation = {
+            **suggestion.evaluation,
+            "status": "failed",
+            "finished_at": timezone.now().isoformat(),
+            "summary": summarize_results(results),
+        }
+        suggestion.save(update_fields=["evaluation"])
+    return True
 
-    The describe check protects rows reclaimed by a live re-trigger of the same deterministic workflow id.
-    """
-    rows = await database_sync_to_async(_list_stale_observations, thread_sensitive=False)()
-    if not rows:
-        return 0
-    temporal = await async_connect()
+
+async def _reap_observations(temporal: Client, rows: list[dict[str, Any]]) -> int:
     reapable, skipped_open, skipped_temporal_error = await classify_stale_rows(
         temporal, rows, workflow_id_key="workflow_id"
     )
+    activity.heartbeat({"phase": "observations_described", "reapable": len(reapable)})
     reaped = 0
     for row in reapable:
         snapshot = row["scanner_snapshot"] or {}
@@ -77,3 +130,31 @@ async def reap_orphaned_observations_activity() -> int:
         skipped_temporal_error=skipped_temporal_error,
     )
     return reaped
+
+
+async def _reap_evaluations(temporal: Client, rows: list[dict[str, Any]]) -> int:
+    reapable, _, _ = await classify_stale_rows(temporal, rows, workflow_id_key="workflow_id")
+    activity.heartbeat({"phase": "evaluations_described", "reapable": len(reapable)})
+    reaped = 0
+    for row in reapable:
+        if await database_sync_to_async(_fail_evaluation, thread_sensitive=False)(row["id"]):
+            reaped += 1
+    logger.info("replay_vision.reap_stuck_evaluations", scanned=len(rows), reaped=reaped)
+    return reaped
+
+
+@activity.defn
+@track_activity()
+async def reap_orphaned_observations_activity() -> int:
+    """Fail pending/running rows past the orphan cutoff whose workflow is no longer open, and settle
+    prompt-suggestion evaluations stuck in `running`; returns the total count reaped.
+
+    The describe check protects rows reclaimed by a live re-trigger of the same deterministic workflow id.
+    """
+    rows = await database_sync_to_async(_list_stale_observations, thread_sensitive=False)()
+    evaluations = await database_sync_to_async(_list_stale_evaluations, thread_sensitive=False)()
+    if not rows and not evaluations:
+        return 0
+    activity.heartbeat({"phase": "listed", "observations": len(rows), "evaluations": len(evaluations)})
+    temporal = await async_connect()
+    return await _reap_observations(temporal, rows) + await _reap_evaluations(temporal, evaluations)
diff --git a/products/replay_vision/backend/temporal/constants.py b/products/replay_vision/backend/temporal/constants.py
index e6ffd4a4d7ce..251932a82c17 100644
--- a/products/replay_vision/backend/temporal/constants.py
+++ b/products/replay_vision/backend/temporal/constants.py
@@ -12,6 +12,8 @@
 # Bounds one reaper pass; a backlog beyond this drains across subsequent reconciler ticks.
 REAP_ORPHANED_OBSERVATIONS_BATCH_SIZE = 500
 REAP_ORPHANED_OBSERVATIONS_TIMEOUT = dt.timedelta(minutes=3)
+# The reaper heartbeats between phases; a pass that goes quiet this long is stalled, not slow.
+REAP_ORPHANED_OBSERVATIONS_HEARTBEAT_TIMEOUT = dt.timedelta(seconds=60)
 
 # Per-action vision-action child, fire-and-forgot by the sweep. Name + timeout live here (not in the
 # workflow-def module) so the sweep can start it without cross-importing another @wf.defn module.
diff --git a/products/replay_vision/backend/temporal/evaluation_types.py b/products/replay_vision/backend/temporal/evaluation_types.py
index 72fccb8fe112..8273d4eb8586 100644
--- a/products/replay_vision/backend/temporal/evaluation_types.py
+++ b/products/replay_vision/backend/temporal/evaluation_types.py
@@ -15,6 +15,10 @@ class EvaluatePromptSuggestionInputs(BaseModel, frozen=True):
     # The edited config the user is testing. None re-runs the stored suggested_config. Defaulted so an
     # in-flight run replaying without this field decodes it as "test the suggestion" (its original behavior).
     config_override: dict[str, Any] | None = None
+    # `evaluation.started_at` as stamped when this run was requested. Usage receipt ids are keyed on it,
+    # so it is frozen into the inputs rather than re-read from the mutable row mid-run. Defaulted so an
+    # in-flight run replaying without this field falls back to re-reading, its original behavior.
+    started_at: str | None = None
 
 
 class EvaluationSession(BaseModel, frozen=True):
@@ -31,6 +35,8 @@ class SelectEvaluationSessionsInputs(BaseModel, frozen=True):
     team_id: int
     session_limit: int | None = None
     config_override: dict[str, Any] | None = None
+    # See `EvaluatePromptSuggestionInputs.started_at`; None restamps, its original behavior.
+    started_at: str | None = None
 
 
 class SelectEvaluationSessionsOutput(BaseModel, frozen=True):
@@ -50,6 +56,8 @@ class RecordEvaluationResultInputs(BaseModel, frozen=True):
     error: str | None = None
     # Defaulted to False so an in-flight run replaying without this field decodes it as a non-preview run.
     preview: bool = False
+    # See `EvaluatePromptSuggestionInputs.started_at`; None falls back to re-reading the row.
+    started_at: str | None = None
 
 
 class FinalizeEvaluationInputs(BaseModel, frozen=True):
diff --git a/products/replay_vision/backend/temporal/evaluation_workflow.py b/products/replay_vision/backend/temporal/evaluation_workflow.py
index 914b9d2887a0..14c6897eb592 100644
--- a/products/replay_vision/backend/temporal/evaluation_workflow.py
+++ b/products/replay_vision/backend/temporal/evaluation_workflow.py
@@ -85,6 +85,7 @@ async def run(self, inputs: EvaluatePromptSuggestionInputs) -> None:
                 team_id=inputs.team_id,
                 session_limit=inputs.session_limit,
                 config_override=inputs.config_override,
+                started_at=inputs.started_at,
             ),
             start_to_close_timeout=dt.timedelta(minutes=1),
             retry_policy=_STATE_RETRY,
@@ -147,6 +148,8 @@ async def _evaluate_session(
                 upload_video_to_gemini_activity,
                 UploadVideoToGeminiInputs(asset_id=asset_result.asset_id),
                 start_to_close_timeout=dt.timedelta(minutes=10),
+                # The activity heartbeats, so a dead worker costs ~2 minutes, not the full budget.
+                heartbeat_timeout=dt.timedelta(minutes=2),
                 retry_policy=_STEP_RETRY,
             )
             call_output: ScannerCallOutput = await wf.execute_activity(
@@ -159,6 +162,7 @@ async def _evaluate_session(
                     snapshot_override=selection.snapshot,
                 ),
                 start_to_close_timeout=dt.timedelta(minutes=10),
+                heartbeat_timeout=dt.timedelta(minutes=2),
                 retry_policy=_STEP_RETRY,
             )
             await self._record(
@@ -199,6 +203,7 @@ async def _record(
                 after_output=after_output,
                 error=error,
                 preview=preview,
+                started_at=inputs.started_at,
             ),
             start_to_close_timeout=dt.timedelta(seconds=30),
             retry_policy=_STATE_RETRY,
diff --git a/products/replay_vision/backend/temporal/reconciler.py b/products/replay_vision/backend/temporal/reconciler.py
index 1250b2176704..221c8ce11e49 100644
--- a/products/replay_vision/backend/temporal/reconciler.py
+++ b/products/replay_vision/backend/temporal/reconciler.py
@@ -14,6 +14,7 @@
 from products.replay_vision.backend.temporal.constants import (
     LIST_ENABLED_SCANNERS_TIMEOUT,
     LIST_SCANNER_SCHEDULES_TIMEOUT,
+    REAP_ORPHANED_OBSERVATIONS_HEARTBEAT_TIMEOUT,
     REAP_ORPHANED_OBSERVATIONS_TIMEOUT,
     REAP_STUCK_VISION_ACTION_RUNS_TIMEOUT,
     RECONCILE_SCHEDULE_OP_TIMEOUT,
@@ -57,6 +58,9 @@ async def run(self, inputs: ReconcileScannerSchedulesInputs) -> ReconcileScanner
             await workflow.execute_activity(
                 reap_orphaned_observations_activity,
                 start_to_close_timeout=REAP_ORPHANED_OBSERVATIONS_TIMEOUT,
+                # The activity heartbeats between phases, so a stalled pass is cut loose before it burns
+                # the whole tick budget.
+                heartbeat_timeout=REAP_ORPHANED_OBSERVATIONS_HEARTBEAT_TIMEOUT,
                 retry_policy=RetryPolicy(maximum_attempts=1),
             )
         except Exception:
diff --git a/products/replay_vision/backend/temporal/scanners/base.py b/products/replay_vision/backend/temporal/scanners/base.py
index bc902428a789..6cc2b784da00 100644
--- a/products/replay_vision/backend/temporal/scanners/base.py
+++ b/products/replay_vision/backend/temporal/scanners/base.py
@@ -169,6 +169,7 @@ def preamble(
         session_metadata: dict[str, Any] | None = None,
         navigation: list[dict[str, Any]] | None = None,
         navigation_dropped: int = 0,
+        events_truncated: bool = False,
     ) -> str:
         """The conversation's shared opening: framing, footer, events tool, calibration, navigation timeline, and
         session metadata. `navigation` takes dumped `NavigationEntry` dicts (plain dicts keep this module free of a
@@ -179,6 +180,7 @@ def preamble(
             session_metadata=session_metadata or {},
             navigation=navigation or [],
             navigation_dropped=navigation_dropped,
+            events_truncated=events_truncated,
         )
 
     def core_steps(self) -> list[MissionStep]:
diff --git a/products/replay_vision/backend/temporal/scanners/prompts/preamble.jinja b/products/replay_vision/backend/temporal/scanners/prompts/preamble.jinja
index b882bba325a5..7b76a37939d1 100644
--- a/products/replay_vision/backend/temporal/scanners/prompts/preamble.jinja
+++ b/products/replay_vision/backend/temporal/scanners/prompts/preamble.jinja
@@ -20,7 +20,9 @@ The replay reconstructs each page from DOM snapshots, so states the user never s
 You have a tool, `get_events_around`: pass a footer `REC_T` and it returns the events within a few seconds of that moment, each tagged with its own `rec_t`. Fields include `event`, `$current_url`, `$event_type`, `elements_chain_*` (what was interacted with), and `$exception_types`/`$exception_values`.
 
 Use it liberally — call it for any moment that looks interesting, and always before you judge whether the user succeeded, failed, or hit friction, which are easy to misread from the video alone. `$rageclick` (the same element clicked again and again) and `$dead_click` (a click that changed nothing) are the highest-signal friction events: treat them as ground truth, concluding one only when `get_events_around` confirms it at that `REC_T`, not from interaction bursts on tabs, dropdowns, or other re-clickable controls.
-</events_tool>
+{% if events_truncated %}
+This session produced more events than the tool can hold, so only the earliest portion is available. Later moments may return nothing even though something happened there — treat a missing result near the end of the session as "unknown", not as "nothing happened".
+{% endif %}</events_tool>
 
 <gestures>
 Touch and browser gestures often leave no click events: a back-swipe navigates without any click, and scroll flicks can register as taps. On touch devices a single `$dead_click` is often a scroll gesture rather than frustration, so weigh it lower than on desktop unless it repeats on the same element. A burst of rapid, aimless taps and swipes with no coherent goal is more likely accidental input (an unlocked phone in a pocket) than a frustrated user.
diff --git a/products/replay_vision/backend/temporal/scanners/scorer.py b/products/replay_vision/backend/temporal/scanners/scorer.py
index 0f3591cea246..68db39fd732a 100644
--- a/products/replay_vision/backend/temporal/scanners/scorer.py
+++ b/products/replay_vision/backend/temporal/scanners/scorer.py
@@ -25,6 +25,11 @@ def _min_lt_max(self) -> "ScoreScale":
         return self
 
 
+# Applied when a stored config carries no scale (legacy or direct-write rows). Lives with the contract so
+# the proposer's grounding and the patch fallback can't drift from what `ScoreScale` accepts.
+DEFAULT_SCORE_SCALE = ScoreScale(min=1.0, max=5.0)
+
+
 class ScorerOutput(BaseScannerOutput, frozen=True):
     scanner_type: Literal[ScannerType.SCORER] = ScannerType.SCORER
     score: float = Field(description="Numeric score on the configured scale.")
diff --git a/products/replay_vision/backend/temporal/types.py b/products/replay_vision/backend/temporal/types.py
index 039afbdb0a30..ad9bcd01a48d 100644
--- a/products/replay_vision/backend/temporal/types.py
+++ b/products/replay_vision/backend/temporal/types.py
@@ -185,6 +185,8 @@ class ScannerLlmInputs(BaseModel, frozen=True):
     # Chronological URL-change timeline rendered into the preamble. Defaults keep pre-existing Redis blobs loadable.
     navigation: list[NavigationEntry] = Field(default_factory=list)
     navigation_dropped: int = Field(default=0, ge=0)
+    # True when the session hit the fetch row cap, so the events tool can't see the whole session.
+    events_truncated: bool = False
     metadata: SessionMetadata
     # Carried for signal emission, not the prompt — kept off `SessionMetadata` so it never reaches the LLM.
     distinct_id: str | None = None
diff --git a/products/replay_vision/backend/temporal/vision_actions/alerts.py b/products/replay_vision/backend/temporal/vision_actions/alerts.py
index 3b0bd1993d17..87dea06c0b82 100644
--- a/products/replay_vision/backend/temporal/vision_actions/alerts.py
+++ b/products/replay_vision/backend/temporal/vision_actions/alerts.py
@@ -76,7 +76,14 @@ def _evaluate(inputs: EvaluateAlertInputs) -> EvaluateAlertResult:
     # Idempotency: a retry after the message was already persisted must re-report FIRED, not re-evaluate
     # against a window that may have shifted.
     if run.synthesized_markdown:
-        return EvaluateAlertResult(status=AlertStatus.FIRED, observation_count=run.observation_count)
+        # `_persist_recovered` writes a message too, so the `recovered` marker is what tells the two
+        # apart. Without it a retried recovery re-reports FIRED and delivers the notification the
+        # RECOVERED branch exists to suppress.
+        recovered = bool((run.output or {}).get("recovered"))
+        return EvaluateAlertResult(
+            status=AlertStatus.RECOVERED if recovered else AlertStatus.FIRED,
+            observation_count=run.observation_count,
+        )
 
     selection: dict[str, Any] = action.selection or {}
     requested_scanner_ids = selection.get("scanner_ids") or ([str(action.scanner_id)] if action.scanner_id else [])
diff --git a/products/replay_vision/backend/temporal/workflow.py b/products/replay_vision/backend/temporal/workflow.py
index 243bf3eee26a..f17321e9fb29 100644
--- a/products/replay_vision/backend/temporal/workflow.py
+++ b/products/replay_vision/backend/temporal/workflow.py
@@ -269,8 +269,11 @@ async def run(self, inputs: ApplyScannerInputs) -> None:
                             exported_asset_id=asset_result.asset_id,
                             signals=call_output.signals,
                         ),
-                        start_to_close_timeout=dt.timedelta(seconds=30),
-                        retry_policy=common.RetryPolicy(maximum_attempts=1),
+                        # 30s truncated the tail on a slow facade and recorded signals_count=0; retries are
+                        # safe because each finding carries a deterministic idempotency key.
+                        start_to_close_timeout=dt.timedelta(minutes=2),
+                        heartbeat_timeout=dt.timedelta(seconds=30),
+                        retry_policy=common.RetryPolicy(maximum_attempts=3),
                     )
                 except Exception:
                     wf.logger.exception("Signal emission activity failed for observation %s", observation_id)
diff --git a/products/replay_vision/backend/tests/test_api.py b/products/replay_vision/backend/tests/test_api.py
index 141d97af40d9..816e01919c71 100644
--- a/products/replay_vision/backend/tests/test_api.py
+++ b/products/replay_vision/backend/tests/test_api.py
@@ -24,6 +24,7 @@
     ObservationTrigger,
     ReplayObservation,
 )
+from products.replay_vision.backend.models.replay_observation_label import ReplayObservationLabel
 from products.replay_vision.backend.models.replay_scanner import (
     ReplayScanner,
     ScannerModel,
@@ -2091,15 +2092,20 @@ def test_retry_keeps_row_when_quota_exhausted(
         self.assertTrue(ReplayObservation.objects.filter(id=observation.id).exists())
         start_workflow.assert_not_called()
 
-    def test_retry_dispatch_failure_returns_503_with_row_restored(
+    def test_retry_dispatch_failure_returns_503_with_row_and_label_restored(
         self, mock_sync_connect: MagicMock, mock_async_to_sync: MagicMock
     ) -> None:
         # The replacement run never started, so the failed row must come back instead of leaving the
-        # recording looking unscanned while the usage ledger still counts the failed attempt.
+        # recording looking unscanned while the usage ledger still counts the failed attempt. The delete
+        # cascades the shared rating away, so restoring only the row silently loses the team's feedback
+        # while the response claims the observation was kept.
         mock_sync_connect.return_value = MagicMock()
         mock_async_to_sync.return_value = MagicMock(side_effect=RuntimeError("temporal unavailable"))
         observation = self._create_failed("sess-broken")
         original_created_at = observation.created_at
+        label = ReplayObservationLabel.objects.create(
+            observation=observation, team=self.team, is_correct=False, feedback="missed the error banner"
+        )
 
         resp = self.client.post(self.retry_url(str(observation.id)))
         self.assertEqual(resp.status_code, 503)
@@ -2108,17 +2114,26 @@ def test_retry_dispatch_failure_returns_503_with_row_restored(
         restored = ReplayObservation.objects.get(id=observation.id)
         self.assertEqual(restored.status, ObservationStatus.FAILED)
         self.assertEqual(restored.created_at, original_created_at)
+        restored_label = ReplayObservationLabel.objects.get(observation_id=observation.id)
+        self.assertEqual(restored_label.id, label.id)
+        self.assertFalse(restored_label.is_correct)
+        self.assertEqual(restored_label.feedback, "missed the error banner")
+        self.assertEqual(restored_label.created_at, label.created_at)
 
-    def test_retry_returns_429_and_restores_row_when_the_atomic_claim_is_refused(
+    def test_retry_returns_429_and_keeps_row_and_label_when_the_atomic_claim_is_refused(
         self, mock_sync_connect: MagicMock, mock_async_to_sync: MagicMock
     ) -> None:
-        # The claim can refuse after the snapshot pre-check passed; the retry must 429 and bring the
-        # failed row back rather than deleting it while no replacement run started.
+        # The claim can refuse after the snapshot pre-check passed. Claiming before the delete keeps a
+        # capped retry a pure no-op: deleting first would cascade away the team's rating on a request
+        # that changes nothing.
         mock_sync_connect.return_value = MagicMock()
         start_workflow = MagicMock()
         mock_async_to_sync.return_value = start_workflow
         observation = self._create_failed("sess-capped")
         original_created_at = observation.created_at
+        ReplayObservationLabel.objects.create(
+            observation=observation, team=self.team, is_correct=False, feedback="missed the error banner"
+        )
 
         with patch("products.replay_vision.backend.api.trigger.try_claim_enqueue_slot", return_value=False):
             resp = self.client.post(self.retry_url(str(observation.id)))
@@ -2128,6 +2143,35 @@ def test_retry_returns_429_and_restores_row_when_the_atomic_claim_is_refused(
         restored = ReplayObservation.objects.get(id=observation.id)
         self.assertEqual(restored.status, ObservationStatus.FAILED)
         self.assertEqual(restored.created_at, original_created_at)
+        self.assertEqual(
+            ReplayObservationLabel.objects.get(observation_id=observation.id).feedback, "missed the error banner"
+        )
+
+    def test_retry_reports_409_when_the_replacement_run_already_holds_the_session(
+        self, mock_sync_connect: MagicMock, mock_async_to_sync: MagicMock
+    ) -> None:
+        # A start we couldn't confirm may still have persisted its own row for this (scanner, session).
+        # Restoring on top of it violates the unique constraint, which used to surface as a 500.
+        observation = self._create_failed("sess-raced")
+
+        def start_and_take_the_slot(*args, **kwargs):
+            ReplayObservation.objects.create(
+                scanner=self.scanner,
+                session_id="sess-raced",
+                scanner_snapshot=_snapshot_for(self.scanner),
+                triggered_by=ObservationTrigger.RETRY,
+                status=ObservationStatus.PENDING,
+            )
+            raise RuntimeError("temporal unavailable")
+
+        mock_sync_connect.return_value = MagicMock()
+        mock_async_to_sync.return_value = MagicMock(side_effect=start_and_take_the_slot)
+
+        resp = self.client.post(self.retry_url(str(observation.id)))
+
+        self.assertEqual(resp.status_code, 409, resp.json())
+        self.assertIn("still finishing", resp.json()["detail"])
+        self.assertEqual(ReplayObservation.objects.filter(scanner=self.scanner, session_id="sess-raced").count(), 1)
 
     def _personal_api_key(self, scopes: list[str]) -> str:
         value = generate_random_token_personal()
diff --git a/products/replay_vision/backend/tests/test_prompt_evaluation.py b/products/replay_vision/backend/tests/test_prompt_evaluation.py
index 4956de18b976..5a872f83ef35 100644
--- a/products/replay_vision/backend/tests/test_prompt_evaluation.py
+++ b/products/replay_vision/backend/tests/test_prompt_evaluation.py
@@ -14,7 +14,7 @@
 )
 from products.replay_vision.backend.models.replay_observation_label import ReplayObservationLabel
 from products.replay_vision.backend.models.replay_observation_usage import ReplayObservationUsage
-from products.replay_vision.backend.models.replay_scanner import ScannerType
+from products.replay_vision.backend.models.replay_scanner import ScannerModel, ScannerType
 from products.replay_vision.backend.models.replay_scanner_prompt_suggestion import (
     ReplayScannerPromptSuggestion,
     SuggestionStatus,
@@ -23,8 +23,10 @@
     EVALUATE_PROMPT_SUGGESTION_EXECUTION_TIMEOUT,
     EVALUATION_SESSION_CAP,
     EVALUATION_SESSION_DEFAULT,
+    build_running_evaluation,
     classify_outcome,
     evaluation_supported,
+    in_flight_evaluation_credits,
     is_preview_evaluation,
     primary_outcome,
     select_evaluation_observations,
@@ -88,7 +90,11 @@ def _create_suggestion(self, **overrides) -> ReplayScannerPromptSuggestion:
         [
             ({"verdict": "Yes "}, "Verdict: yes"),
             ({"verdict": "no", "tags": ["a"]}, "Verdict: no"),
-            ({"tags": ["Churn ", "bug", "churn"]}, "Tags: bug, churn, churn"),
+            # Slug-normalized and deduped, so casing and spacing can't read as a changed outcome.
+            ({"tags": ["Churn ", "bug", "churn"]}, "Tags: bug, churn"),
+            # Freeform tags count too: a rewrite that only moves those must not evaluate as "no change".
+            ({"tags": ["bug"], "tags_freeform": ["Payment Issues"]}, "Tags: bug, payment_issues"),
+            ({"tags": [], "tags_freeform": ["checkout"]}, "Tags: checkout"),
             # Preview types: scorer shows the raw score, summarizer prefers the title then falls back to the summary.
             ({"score": 7}, "Score: 7"),
             ({"score": 3.5}, "Score: 3.5"),
@@ -430,6 +436,11 @@ def test_evaluate_starts_workflow_and_stamps_running(self) -> None:
         self.assertEqual(client.start_workflow.await_args.args[1].session_limit, EVALUATION_SESSION_DEFAULT)
         # No edited config posted, so the run tests the stored suggestion.
         self.assertIsNone(client.start_workflow.await_args.args[1].config_override)
+        # Receipt ids are keyed on started_at, so the run must carry the stamp it was started with rather
+        # than re-reading a row a concurrent re-test can restamp underneath it.
+        suggestion.refresh_from_db()
+        assert suggestion.evaluation is not None
+        self.assertEqual(client.start_workflow.await_args.args[1].started_at, suggestion.evaluation["started_at"])
 
     def test_evaluate_passes_edited_config_to_workflow(self) -> None:
         self._create_rated()
@@ -575,6 +586,26 @@ def test_running_evaluation_elsewhere_counts_against_quota(self) -> None:
         self.assertEqual(resp.status_code, 402)
         client.start_workflow.assert_not_awaited()
 
+    def test_in_flight_reservation_prices_from_the_frozen_model(self) -> None:
+        # Receipts bill the model frozen at workflow start. Pricing the reservation from the scanner's
+        # current model instead lets an edit mid-run silently re-price committed spend.
+        expensive, cheap = ScannerModel.GEMINI_3_6_FLASH, ScannerModel.GEMINI_3_5_FLASH_LITE
+        scanner = self._create_scanner(name="frozen-model", model=expensive)
+        ReplayScannerPromptSuggestion.objects.create(
+            scanner=scanner,
+            team=self.team,
+            suggested_prompt="p",
+            status=SuggestionStatus.PENDING,
+            scanner_version=1,
+            evaluation=build_running_evaluation(total=3, labels_fingerprint="", model=expensive),
+        )
+        reserved = in_flight_evaluation_credits(self.team.organization_id)
+
+        scanner.model = cheap
+        scanner.save(update_fields=["model"])
+        self.assertEqual(in_flight_evaluation_credits(self.team.organization_id), reserved)
+        self.assertEqual(reserved, 3 * observation_credits_for_model(expensive))
+
     @parameterized.expand([("zero", 0), ("above_cap", EVALUATION_SESSION_CAP + 1)])
     def test_evaluate_rejects_out_of_range_session_limit(self, _name: str, limit: int) -> None:
         self._create_rated()
diff --git a/products/replay_vision/backend/tests/test_proposers.py b/products/replay_vision/backend/tests/test_proposers.py
index 777651cb7382..fa2b78b17437 100644
--- a/products/replay_vision/backend/tests/test_proposers.py
+++ b/products/replay_vision/backend/tests/test_proposers.py
@@ -128,12 +128,20 @@ def test_classifier_patch_applies_tag_ops() -> None:
         {"op": "rename", "tag": "missing", "to": "renamed"},  # nothing to rename
         {"op": "add"},  # malformed: no tag
         {"tag": "orphan"},  # malformed: no op
+        "payment_failed",  # malformed: a bare string, not a dict
+        ["add", "payment_failed"],  # malformed: a list
+        {"op": "add", "tag": 42},  # malformed: non-string tag
+        {"op": "add", "tag": "   "},  # malformed: blank tag
+        {"op": "add", "tag": "!!!"},  # slugifies to nothing, so it could never be applied
+        {"op": "rename", "tag": "checkout", "to": 7},  # malformed: non-string destination
+        {"op": "drop", "tag": "checkout"},  # not one of the three operations
+        {"op": "add", "tag": "Checkout"},  # same slug as an existing tag, so uniqueness would reject it
     ],
 )
-def test_classifier_patch_ignores_ops_that_dont_apply(op: dict[str, Any]) -> None:
+def test_classifier_patch_ignores_ops_that_dont_apply(op: Any) -> None:
     # An LLM tag op can reference a tag that's already gone/present/never existed, or be malformed (a model
-    # response the schema should have rejected). _apply_tag_ops must leave the vocabulary untouched rather
-    # than duplicate a tag or raise a KeyError (which would 500 the whole generation instead of a usable one).
+    # response the schema should have rejected). This runs outside the generation fallbacks, so anything that
+    # raises here loses the whole suggestion — including an otherwise usable rewritten prompt.
     proposer = get_proposer("classifier")
     base = {"prompt": "p", "tags": ["checkout"]}
     llm = {"suggested_prompt": "p", "tag_ops": [op], "rationale": "r"}
@@ -145,6 +153,21 @@ def test_classifier_patch_ignores_ops_that_dont_apply(op: dict[str, Any]) -> Non
     assert proposer.to_changes(base, suggested, llm) == []
 
 
+@pytest.mark.parametrize("tag_ops", [None, "add checkout", {"op": "add", "tag": "x"}, 5])
+def test_classifier_patch_survives_tag_ops_that_arent_a_list(tag_ops: Any) -> None:
+    # Iterating a dict yields its keys and a string yields characters, so an unguarded loop would either
+    # silently apply nonsense or raise and lose the suggestion.
+    proposer = get_proposer("classifier")
+    base = {"prompt": "p", "tags": ["checkout"]}
+    llm = {"suggested_prompt": "p2", "tag_ops": tag_ops, "rationale": "r"}
+
+    suggested = proposer.to_config_patch(llm, base)
+
+    assert suggested["tags"] == ["checkout"]
+    assert suggested["prompt"] == "p2"
+    assert [c.kind for c in proposer.to_changes(base, suggested, llm)] == ["prompt"]
+
+
 @pytest.mark.parametrize(
     "existing,to,expected",
     [
@@ -199,6 +222,9 @@ def test_summarizer_patch_defends_against_missing_or_invalid_length() -> None:
         ("monitor", {"prompt": "p", "allow_inconclusive": True}, "allow_inconclusive: true"),
         ("scorer", {"prompt": "p", "scale": {"min": 1, "max": 5, "label": "frustration"}}, "1 to 5 (frustration)"),
         ("scorer", {"prompt": "p", "scale": {"min": 0, "max": 10}}, "0 to 10."),
+        # No stored scale: say which one applies, or the model invents a range and the patch records it
+        # as a deliberate change complete with a rationale.
+        ("scorer", {"prompt": "p"}, "1.0 to 5.0"),
         ("summarizer", {"prompt": "p"}, "length: medium"),
         ("summarizer", {"prompt": "p", "length": "long"}, "length: long"),
     ],
diff --git a/products/replay_vision/backend/tests/test_reconciler.py b/products/replay_vision/backend/tests/test_reconciler.py
index eabe4d1cdbb2..91ad27e0b5df 100644
--- a/products/replay_vision/backend/tests/test_reconciler.py
+++ b/products/replay_vision/backend/tests/test_reconciler.py
@@ -15,6 +15,7 @@
 from temporalio.common import SearchAttributePair, TypedSearchAttributes
 from temporalio.exceptions import ApplicationError
 from temporalio.service import RPCError, RPCStatusCode
+from temporalio.testing import ActivityEnvironment
 
 from posthog.models import Organization, Team
 from posthog.temporal.common.search_attributes import POSTHOG_SCHEDULE_FINGERPRINT_KEY
@@ -25,7 +26,12 @@
     ReplayObservation,
 )
 from products.replay_vision.backend.models.replay_scanner import ReplayScanner, ScannerModel, ScannerType
+from products.replay_vision.backend.models.replay_scanner_prompt_suggestion import (
+    ReplayScannerPromptSuggestion,
+    SuggestionStatus,
+)
 from products.replay_vision.backend.models.vision_action import VisionAction, VisionActionRun, VisionActionRunStatus
+from products.replay_vision.backend.prompt_evaluation import EVALUATE_PROMPT_SUGGESTION_EXECUTION_TIMEOUT
 from products.replay_vision.backend.temporal.activities import (
     delete_scanner_schedule_activity,
     list_enabled_scanners_activity,
@@ -44,6 +50,7 @@
     SCANNER_SCHEDULE_ID_PREFIX,
     SWEEP_SCANNER_WORKFLOW_NAME,
     VISION_ACTION_RUN_STUCK_CUTOFF,
+    build_evaluate_prompt_suggestion_workflow_id,
 )
 from products.replay_vision.backend.temporal.reconciler import (
     ReconcileScannerSchedulesWorkflow,
@@ -529,7 +536,7 @@ def _setup() -> dict[str, ReplayObservation]:
         "products.replay_vision.backend.temporal.activities.reap_orphaned_observations.async_connect",
         AsyncMock(return_value=temporal),
     ):
-        reaped = await reap_orphaned_observations_activity()
+        reaped = await ActivityEnvironment().run(reap_orphaned_observations_activity)
 
     assert reaped == 3
     statuses = {
@@ -546,6 +553,61 @@ def _setup() -> dict[str, ReplayObservation]:
     assert set(temporal.described) == {"wf-gone-1", "wf-timed-out", "wf-open", "wf-err"}
 
 
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_reap_settles_stuck_prompt_suggestion_evaluations(org_team) -> None:
+    # The evaluation workflow swallows finalize failures, so a terminated run leaves the row claiming to
+    # be running forever: the UI polls a test that will never finish and quota reserves its unspent credits.
+    _, team = org_team
+    scanner = await sync_to_async(_make_scanner)(team)
+    stale = timezone.now() - (EVALUATE_PROMPT_SUGGESTION_EXECUTION_TIMEOUT * 2 + dt.timedelta(minutes=5))
+
+    def _setup() -> dict[str, ReplayScannerPromptSuggestion]:
+        rows = {}
+        for key, started_at in (("stuck", stale), ("recent", timezone.now()), ("still_open", stale)):
+            rows[key] = ReplayScannerPromptSuggestion.objects.create(
+                scanner=scanner,
+                team=team,
+                suggested_prompt=f"p-{key}",
+                status=SuggestionStatus.PENDING,
+                scanner_version=1,
+                evaluation={
+                    "status": "running",
+                    "started_at": started_at.isoformat(),
+                    "total": 2,
+                    "results": [{"outcome": "kept"}],
+                    "summary": None,
+                },
+            )
+        return rows
+
+    rows = await sync_to_async(_setup)()
+    outcomes = {
+        build_evaluate_prompt_suggestion_workflow_id(rows["stuck"].id): "not_found",
+        build_evaluate_prompt_suggestion_workflow_id(rows["still_open"].id): "open",
+    }
+    temporal = _StubReapTemporal(outcomes)
+
+    with patch(
+        "products.replay_vision.backend.temporal.activities.reap_orphaned_observations.async_connect",
+        AsyncMock(return_value=temporal),
+    ):
+        reaped = await ActivityEnvironment().run(reap_orphaned_observations_activity)
+
+    assert reaped == 1
+    settled = await sync_to_async(lambda: ReplayScannerPromptSuggestion.objects.get(pk=rows["stuck"].pk))()
+    assert settled.evaluation is not None
+    assert settled.evaluation["status"] == "failed"
+    assert settled.evaluation["finished_at"] is not None
+    assert settled.evaluation["summary"] == {"kept": 1, "regressed": 0, "fixed": 0, "still_wrong": 0, "errors": 0}
+    for key in ("recent", "still_open"):
+        untouched = await sync_to_async(lambda k=key: ReplayScannerPromptSuggestion.objects.get(pk=rows[k].pk))()
+        assert untouched.evaluation is not None
+        assert untouched.evaluation["status"] == "running", key
+    # A stamp inside the timeout is never described: only provably-dead runs are settled.
+    assert build_evaluate_prompt_suggestion_workflow_id(rows["recent"].id) not in temporal.described
+
+
 def _make_run(action: VisionAction, *, status: str, workflow_id: str, age: dt.timedelta) -> VisionActionRun:
     run = VisionActionRun.all_teams.create(
         vision_action=action,
diff --git a/products/replay_vision/backend/tests/test_temporal.py b/products/replay_vision/backend/tests/test_temporal.py
index 3b0a6bc238d2..07c9026067d2 100644
--- a/products/replay_vision/backend/tests/test_temporal.py
+++ b/products/replay_vision/backend/tests/test_temporal.py
@@ -1370,7 +1370,7 @@ async def test_session_metadata_round_trips_to_payload(self) -> None:
 
     @pytest.mark.asyncio
     async def test_loads_every_page_until_source_is_exhausted(self) -> None:
-        # No event cap: when a page reports `has_more`, keep paging and load the whole session.
+        # Below the row cap, a page reporting `has_more` keeps paging and loads the whole session.
         scanner = await sync_to_async(_make_scanner)()
         observation_id = uuid.uuid4()
         start = dt.datetime(2026, 5, 12, 10, 0, 0, tzinfo=dt.UTC)
@@ -1398,6 +1398,46 @@ async def test_loads_every_page_until_source_is_exhausted(self) -> None:
         assert stored is not None
         assert mock_obj.get_events.call_count == 2  # paged through both, not capped at one page
         assert len(stored.events.rows) == 3  # every event from both pages
+        assert stored.events_truncated is False
+
+    @pytest.mark.asyncio
+    async def test_stops_paging_at_the_row_cap_and_marks_the_result_truncated(self) -> None:
+        # Eligibility caps active seconds, not event count, so an instrumentation loop can page forever
+        # and blow up worker memory, the Redis blob, and the events index. The prompt has to say so, or
+        # the model reads a missing late event as "nothing happened".
+        scanner = await sync_to_async(_make_scanner)()
+        observation_id = uuid.uuid4()
+        start = dt.datetime(2026, 5, 12, 10, 0, 0, tzinfo=dt.UTC)
+        metadata = {"start_time": start, "end_time": start, "duration": 60, "active_seconds": 30}
+        cols = ["event", "timestamp", "$session_id"]
+
+        # Distinct per page: the cap counts raw rows, before `_process_events` dedups them.
+        def _page(offset: int) -> list[tuple]:
+            return [("$autocapture", start, f"s-{offset + i}") for i in range(4)]
+
+        mock_obj = self._make_session_replay_events_mock(
+            metadata, [(cols, _page(0), True), (cols, _page(4), True), (cols, _page(8), True)]
+        )
+
+        with (
+            patch(
+                "products.replay_vision.backend.temporal.activities.fetch_session_events.SessionReplayEvents",
+                return_value=mock_obj,
+            ),
+            patch("products.replay_vision.backend.temporal.activities.fetch_session_events._MAX_TOTAL_EVENT_ROWS", 6),
+        ):
+            await fetch_session_events_activity(
+                FetchSessionEventsInputs(observation_id=observation_id, team_id=scanner.team_id, session_id="sess-1")
+            )
+
+        redis_client = get_async_client(settings.REPLAY_VISION_REDIS_URL)
+        key = generate_state_key(label=StateActivitiesEnum.SESSION_EVENTS, state_id=str(observation_id))
+        stored = await get_data_class_from_redis(redis_client, key, target_class=ScannerLlmInputs)
+        assert stored is not None
+        # Two pages of 4 reach 8 rows, past the cap of 6: it stops there rather than draining the source.
+        assert mock_obj.get_events.call_count == 2
+        assert len(stored.events.rows) == 6
+        assert stored.events_truncated is True
 
     @pytest.mark.asyncio
     async def test_stops_paging_once_no_more(self) -> None:
diff --git a/products/replay_vision/backend/tests/test_vision_action_alerts.py b/products/replay_vision/backend/tests/test_vision_action_alerts.py
index 294d568fff41..aebd4ff4160c 100644
--- a/products/replay_vision/backend/tests/test_vision_action_alerts.py
+++ b/products/replay_vision/backend/tests/test_vision_action_alerts.py
@@ -166,6 +166,10 @@ def test_recovery_bookends_the_breach_and_rearms_the_alert(self) -> None:
         self.assertIn("below the threshold of 1", recovery_run.synthesized_markdown)
         self.assertIn("had been firing since", recovery_run.synthesized_markdown)
         self.assertTrue(recovery_run.output["recovered"])
+        # A lost attempt after the recovery message committed re-enters the idempotency guard. Reporting
+        # FIRED there would deliver the alert notification that recovering exists to suppress.
+        replayed = _evaluate(EvaluateAlertInputs(run_id=recovery_run.id, team_id=self.team.id))
+        self.assertEqual(replayed.status, AlertStatus.RECOVERED)
         self._record(recovery_run, VisionActionRunStatus.COMPLETED)
 
         # A fresh match after recovery is a new incident — it must fire, not report still_breached.
diff --git a/products/replay_vision/frontend/components/ObservationsDock.tsx b/products/replay_vision/frontend/components/ObservationsDock.tsx
index 5ade67a18d93..1e34a5ee432a 100644
--- a/products/replay_vision/frontend/components/ObservationsDock.tsx
+++ b/products/replay_vision/frontend/components/ObservationsDock.tsx
@@ -70,6 +70,7 @@ function ScannerPicker({ sessionId }: { sessionId: string }): JSX.Element {
                                     fullWidth
                                     size="small"
                                     onClick={() => observe(scanner.id)}
+                                    disabledReason={observing ? 'Starting an observation…' : undefined}
                                     data-attr="vision-scan-pick-scanner"
                                     data-ph-capture-attribute-scanner-type={scanner.scanner_type}
                                 >
diff --git a/products/replay_vision/frontend/generated/api.schemas.ts b/products/replay_vision/frontend/generated/api.schemas.ts
index d4fe4585cea6..e08306914b14 100644
--- a/products/replay_vision/frontend/generated/api.schemas.ts
+++ b/products/replay_vision/frontend/generated/api.schemas.ts
@@ -384,6 +384,14 @@ export interface RunActionResponseApi {
     already_running: boolean
 }
 
+/**
+ * The shape every Replay Vision error response uses, so generated clients read one key.
+ */
+export interface ReplayVisionErrorApi {
+    /** Human-readable explanation of why the request was refused. */
+    detail: string
+}
+
 /**
  * * `running` - Running
  * * `completed` - Completed
diff --git a/products/replay_vision/frontend/logics/observationsDockLogic.test.ts b/products/replay_vision/frontend/logics/observationsDockLogic.test.ts
new file mode 100644
index 000000000000..dc9d6171d5ad
--- /dev/null
+++ b/products/replay_vision/frontend/logics/observationsDockLogic.test.ts
@@ -0,0 +1,54 @@
+import { expectLogic } from 'kea-test-utils'
+
+import { useMocks } from '~/mocks/jest'
+import { initKeaTests } from '~/test/init'
+
+import { observationsDockLogic } from './observationsDockLogic'
+
+describe('observationsDockLogic', () => {
+    let logic: ReturnType<typeof observationsDockLogic.build>
+    let observeCalls: number
+    let releaseObserve: () => void
+
+    beforeEach(() => {
+        observeCalls = 0
+        useMocks({
+            get: {
+                '/api/projects/:team/vision/scanners/': () => [200, { results: [] }],
+                '/api/projects/:team/vision/observations/': () => [200, { results: [] }],
+            },
+            post: {
+                '/api/projects/:team/vision/scanners/:id/observe/': async () => {
+                    observeCalls += 1
+                    // Hold the request open so a second click lands while the first is still in flight.
+                    await new Promise<void>((resolve) => {
+                        releaseObserve = resolve
+                    })
+                    return [202, { workflow_id: 'wf-1' }]
+                },
+            },
+        })
+        initKeaTests()
+        logic = observationsDockLogic({ sessionId: 'sess-1' })
+        logic.mount()
+    })
+
+    afterEach(() => {
+        releaseObserve?.()
+        logic?.unmount()
+    })
+
+    it('starts one observation when the same scanner row is clicked twice', async () => {
+        // The picker rows disable while observing, but both click events can land before React re-renders,
+        // and the duplicate-scanner guard can't see a run that has no observation row yet. Two POSTs mean a
+        // second, contradictory toast on a request the backend refuses anyway.
+        logic.actions.observe('scanner-1')
+        logic.actions.observe('scanner-1')
+        await expectLogic(logic).toMatchValues({ observing: true })
+        // Let both listeners reach (or skip) their request before counting.
+        await new Promise((resolve) => setTimeout(resolve, 0))
+        await new Promise((resolve) => setTimeout(resolve, 0))
+
+        expect(observeCalls).toBe(1)
+    })
+})
diff --git a/products/replay_vision/frontend/logics/observationsDockLogic.ts b/products/replay_vision/frontend/logics/observationsDockLogic.ts
index f275ed766bf0..323b4e6bbfb0 100644
--- a/products/replay_vision/frontend/logics/observationsDockLogic.ts
+++ b/products/replay_vision/frontend/logics/observationsDockLogic.ts
@@ -201,7 +201,10 @@ export const observationsDockLogic = kea<observationsDockLogicType>([
             )
         }
         return {
-            loadObservations: async () => {
+            loadObservations: async (_, breakpoint) => {
+                // Poll, observe, retry, and the SSE hook all call this; without it a slow earlier response
+                // lands last and resurrects a settled card.
+                await breakpoint(1)
                 const teamId = teamLogic.values.currentTeamId
                 if (!teamId) {
                     actions.loadObservationsFailure() // Clear the loading flag; a bare return spins forever.
@@ -209,6 +212,7 @@ export const observationsDockLogic = kea<observationsDockLogicType>([
                 }
                 try {
                     const response = await visionObservationsList(String(teamId), { session_id: props.sessionId })
+                    breakpoint()
                     actions.loadObservationsSuccess(response.results ?? [])
                 } catch {
                     metricCount('replay_vision_frontend_observations_load_failures')
@@ -221,6 +225,11 @@ export const observationsDockLogic = kea<observationsDockLogicType>([
             loadObservationsFailure: reschedulePoll,
 
             observe: async ({ scannerId }) => {
+                // A cache flag, not `values.observing`: the reducer has already flipped that to true by the
+                // time this listener runs, so it would reject every call.
+                if (cache.observeInFlight) {
+                    return
+                }
                 actions.setScannerPickerOpen(false)
                 const teamId = teamLogic.values.currentTeamId
                 if (!teamId) {
@@ -234,6 +243,7 @@ export const observationsDockLogic = kea<observationsDockLogicType>([
                     actions.setDockOpen(true)
                     return
                 }
+                cache.observeInFlight = true
                 try {
                     await visionScannersObserveCreate(String(teamId), scannerId, { session_id: props.sessionId })
                     lemonToast.success('Observation started')
@@ -247,6 +257,8 @@ export const observationsDockLogic = kea<observationsDockLogicType>([
                     metricCount('replay_vision_frontend_observe_failures')
                     lemonToast.error(`Failed to start observation${error.detail ? `: ${error.detail}` : ''}`)
                     actions.observeFailure()
+                } finally {
+                    cache.observeInFlight = false
                 }
             },
 
diff --git a/products/replay_vision/frontend/observations/observationLabelLogic.ts b/products/replay_vision/frontend/observations/observationLabelLogic.ts
index fa38cbba0813..100ada8da96e 100644
--- a/products/replay_vision/frontend/observations/observationLabelLogic.ts
+++ b/products/replay_vision/frontend/observations/observationLabelLogic.ts
@@ -128,6 +128,7 @@ export const observationLabelLogic = kea<observationLabelLogicType>([
 
         rate: async ({ isCorrect, feedback }) => {
             cache.labelEpoch = (cache.labelEpoch ?? 0) + 1
+            const epoch = cache.labelEpoch
             const teamId = teamLogic.values.currentTeamId
             if (!teamId) {
                 return
@@ -138,6 +139,10 @@ export const observationLabelLogic = kea<observationLabelLogicType>([
                     is_correct: isCorrect,
                     feedback,
                 })
+                // Thumbs-up then thumbs-down can settle out of order and leave the older rating on screen.
+                if ((cache.labelEpoch ?? 0) !== epoch) {
+                    return
+                }
                 actions.labelUpdated(label)
                 props.onChange?.(label)
             } catch (error: any) {
@@ -149,6 +154,7 @@ export const observationLabelLogic = kea<observationLabelLogicType>([
 
         clearRating: async () => {
             cache.labelEpoch = (cache.labelEpoch ?? 0) + 1
+            const epoch = cache.labelEpoch
             const teamId = teamLogic.values.currentTeamId
             if (!teamId) {
                 return
@@ -156,6 +162,9 @@ export const observationLabelLogic = kea<observationLabelLogicType>([
             actions.setSaving(true)
             try {
                 await visionObservationsLabelDestroy(String(teamId), props.observationId)
+                if ((cache.labelEpoch ?? 0) !== epoch) {
+                    return
+                }
                 actions.labelUpdated(null)
                 props.onChange?.(null)
             } catch (error: any) {
diff --git a/products/replay_vision/frontend/observations/replayObservationLogic.ts b/products/replay_vision/frontend/observations/replayObservationLogic.ts
index 30d5c2de73c2..bcc615eb5597 100644
--- a/products/replay_vision/frontend/observations/replayObservationLogic.ts
+++ b/products/replay_vision/frontend/observations/replayObservationLogic.ts
@@ -132,6 +132,7 @@ export const replayObservationLogic = kea<replayObservationLogicType>([
             loadObservation: async () => {
                 const teamId = teamLogic.values.currentTeamId
                 if (!teamId) {
+                    actions.loadObservationFailure() // Clear the loading flag; a bare return spins forever.
                     return
                 }
                 try {
diff --git a/products/replay_vision/frontend/replay_scanners/scannerCalibrationLogic.ts b/products/replay_vision/frontend/replay_scanners/scannerCalibrationLogic.ts
index 27a9983d32ea..9138aa032a40 100644
--- a/products/replay_vision/frontend/replay_scanners/scannerCalibrationLogic.ts
+++ b/products/replay_vision/frontend/replay_scanners/scannerCalibrationLogic.ts
@@ -525,6 +525,7 @@ export const scannerCalibrationLogic = kea<scannerCalibrationLogicType>([
         loadObservations: async (_, breakpoint) => {
             const teamId = teamLogic.values.currentTeamId
             if (!teamId) {
+                actions.loadObservationsFailure() // Clear the loading flag; a bare return spins forever.
                 return
             }
             let response: Awaited<ReturnType<typeof visionScannersObservationsList>>
@@ -580,6 +581,7 @@ export const scannerCalibrationLogic = kea<scannerCalibrationLogicType>([
         loadLabelStats: async (_, breakpoint) => {
             const teamId = teamLogic.values.currentTeamId
             if (!teamId) {
+                actions.loadLabelStatsFailure()
                 return
             }
             let response
@@ -598,6 +600,7 @@ export const scannerCalibrationLogic = kea<scannerCalibrationLogicType>([
         loadCurrentSuggestion: async (_, breakpoint) => {
             const teamId = teamLogic.values.currentTeamId
             if (!teamId) {
+                actions.loadCurrentSuggestionFailure()
                 return
             }
             const epoch = cache.suggestionEpoch ?? 0
diff --git a/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.test.ts b/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.test.ts
index c420e3420485..12cfd6828ca6 100644
--- a/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.test.ts
+++ b/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.test.ts
@@ -52,7 +52,7 @@ describe('scannerRunTabLogic', () => {
         logic?.unmount()
     })
 
-    it('keeps the newest observation per session and does not cap the page to the visible-row count', async () => {
+    it('keeps the newest observation per session and leaves retry headroom above the visible-row count', async () => {
         await expectLogic(logic, () => logic.actions.setVisibleSessionIds(['s1', 's2', 's3'])).toDispatchActions([
             'loadObservationsSuccess',
         ])
@@ -71,7 +71,9 @@ describe('scannerRunTabLogic', () => {
         // The connected replayScannerLogic fires its own paged list load; ours is the session_id lookup.
         const lookupUrl = requestedUrls.find((url) => url.includes('session_id='))
         expect(lookupUrl).not.toBeUndefined()
-        expect(lookupUrl).not.toContain('limit=')
+        // On the server's default page size the tail is dropped and those sessions render "Not scanned";
+        // one-per-row would drop them as soon as a retry stacks a second observation onto a session.
+        expect(lookupUrl).toContain('limit=12')
     })
 
     it('releases the pending bridge once the scanned session lands in the lookup', async () => {
diff --git a/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.ts b/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.ts
index 4da291d477be..d7274e23d4bc 100644
--- a/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.ts
+++ b/products/replay_vision/frontend/replay_scanners/scannerRunTabLogic.ts
@@ -29,6 +29,9 @@ export interface RowObservation {
 
 export const IN_PROGRESS_STATUSES = new Set<string>(['pending', 'running'])
 
+// Headroom per visible session for the observations a retry stacks on top of the original scan.
+const OBSERVATIONS_PER_SESSION_ALLOWANCE = 4
+
 export interface ScannerRunTabLogicProps {
     scannerId: string
 }
@@ -259,11 +262,13 @@ export const scannerRunTabLogic = kea<scannerRunTabLogicType>([
                     return
                 }
                 try {
-                    // No limit coupling to visible rows — retries stack observations and a truncated page hides scans.
                     // Ordering pinned: the newest-wins mapping below depends on it, not on the API default.
+                    // Without a limit the server's default page truncates the tail and those sessions read
+                    // as "Not scanned" when they have been.
                     const response = await visionScannersObservationsList(String(teamId), props.scannerId, {
                         session_id: sessionIds.join(','),
                         order_by: '-created_at',
+                        limit: sessionIds.length * OBSERVATIONS_PER_SESSION_ALLOWANCE,
                     })
                     breakpoint()
                     const bySession: Record<string, RowObservation> = {}
diff --git a/products/replay_vision/mcp/tools.yaml b/products/replay_vision/mcp/tools.yaml
index ebbe4db8dc95..73cb598f8edd 100644
--- a/products/replay_vision/mcp/tools.yaml
+++ b/products/replay_vision/mcp/tools.yaml
@@ -40,7 +40,7 @@ tools:
             - created_by
             - updated_at
         requires_ai_consent: true
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-delete:
         operation: vision_actions_destroy
         enabled: true
@@ -55,7 +55,7 @@ tools:
             Permanently delete a vision action by ID, along with its run history (past group summaries and alert
             firings). Its delivery destinations stop firing. To stop an action without losing its history, set `enabled:
             false` via `vision-actions-update` instead.
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-list:
         operation: vision_actions_list
         enabled: true
@@ -74,7 +74,7 @@ tools:
             alert` delivers only when its alert condition holds. Each row carries the bound `scanner`, `mode`,
             `trigger_type`, cadence, and `is_scanner_digest` (the built-in daily digest). Pass `?scanner=<id>` to filter
             to one scanner. To read a specific action's produced reports, list its runs with `vision-actions-runs-list`.
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-retrieve:
         operation: vision_actions_retrieve
         enabled: true
@@ -90,7 +90,7 @@ tools:
             (group_summary/alert), `selection` (the observation-targeting predicate — verdict/tags/score bounds/window),
             `synthesis_config`, `alert_config`, cadence, and delivery targets. Use `vision-actions-runs-list` +
             `vision-actions-runs-retrieve` to read the reports it has actually produced.
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-run-create:
         operation: vision_actions_run_create
         enabled: false
@@ -112,7 +112,7 @@ tools:
             (running/completed/failed/skipped), `scheduled_at`, `observation_count` (how many observations fed the
             summary), `error_reason`, and `is_recovery` — without the report body. Fetch a completed run with
             `vision-actions-runs-retrieve` to get the synthesized report and the observations it cited.
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-runs-retrieve:
         operation: vision_actions_runs_retrieve
         enabled: true
@@ -132,7 +132,7 @@ tools:
             markdown, `observations[N-1]` is the observation it cites, so you can check whether the cited observation
             actually supports the claim. Empty `synthesized_markdown`/`observations` means the run skipped, failed, or
             predates observation tracking.
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-actions-update:
         operation: vision_actions_partial_update
         enabled: true
@@ -160,7 +160,7 @@ tools:
             - created_by
             - updated_at
         requires_ai_consent: true
-        feature_flag: replay-vision
+        feature_flag: replay-vision-actions
     vision-observations-create-task-create:
         operation: vision_observations_create_task_create
         enabled: false
@@ -254,9 +254,10 @@ tools:
             model). Returns `credit_limit` (null when uncapped), `credits_used` (in-flight + succeeded observations
             counted against it), `remaining`, `exhausted`, `projected_monthly_credits`, and the
             `period_start`/`period_end` of the current calendar-month window. Check `remaining`/`exhausted` before
-            creating a scanner or calling `vision-scanners-scan-session` — observations requested over budget are
-            silently skipped until the next period. Pair with `vision-scanners-estimate-create` to size a new scanner
-            against the remaining budget.
+            creating a scanner or calling `vision-scanners-scan-session`: over budget, a scanner's scheduled sweeps
+            silently skip their observations until the next period, while `vision-scanners-scan-session` is rejected
+            outright with a 402. Pair with `vision-scanners-estimate-create` to size a new scanner against the remaining
+            budget.
         feature_flag: replay-vision
     vision-scanners-affected-cohort-create:
         operation: vision_scanners_affected_cohort_create
@@ -292,18 +293,18 @@ tools:
         title: Create replay vision scanner
         description: >
             Create a replay vision scanner. Pick a `scanner_type` — `monitor` (open-ended observations), `classifier`
-            (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary with
-            optional facet embeddings via `emits_embeddings`) — and fill `scanner_config` to match: all types need a
-            `prompt`; classifiers also need `tags`; scorers also need `scale`; summarizers optionally set `length` and
-            `emits_embeddings`. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches —
+            (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, which
+            always emits facet embeddings) — and fill `scanner_config` to match: all types need a `prompt`; classifiers
+            also need `tags`; scorers also need `scale`; summarizers optionally set `length`. Unknown `scanner_config`
+            keys are rejected. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches —
             `date_from` and `date_to` are ignored (the schedule controls time). `sampling_rate` is a 0..1 random
-            downsample applied after the query matches (default 1.0). `model` chooses the LLM. Names must be unique
-            within the team. Before creating a broad scanner (a permissive `query` and/or high `sampling_rate`), first
-            call `vision-scanners-estimate-create` to project its monthly observation volume and `vision-quota-retrieve`
-            to check the org's remaining budget — an enabled scanner sweeps matching recordings every 5 minutes, and
-            creation itself does not check quota, so a too-broad scanner can exhaust the monthly budget on its first
-            sweeps. To get notified about what the scanner finds (a Slack alert or a scheduled summary), create a vision
-            action bound to it with `vision-actions-create`.
+            downsample applied after the query matches (default 1.0). `model` chooses the LLM, and sets what each
+            observation costs in credits. Names must be unique within the team. Before creating a broad scanner (a
+            permissive `query` and/or high `sampling_rate`), first call `vision-scanners-estimate-create` to project its
+            monthly credit spend and `vision-quota-retrieve` to check the org's remaining budget — an enabled scanner
+            sweeps matching recordings every 5 minutes, and creation itself does not check quota, so a too-broad scanner
+            can exhaust the monthly budget on its first sweeps. To get notified about what the scanner finds (a Slack
+            alert or a scheduled summary), create a vision action bound to it with `vision-actions-create`.
         exclude_params:
             - id
             - scanner_version
@@ -346,10 +347,13 @@ tools:
         description: >
             Preview how many observations a proposed scanner would generate before saving it. Send a `query`
             (`RecordingsQuery` shape; `date_from`/`date_to` are ignored — the estimate uses a fixed 30-day lookback) and
-            an optional `sampling_rate`. Returns `matched_sessions_in_window`, the `window_days` it was measured over,
-            and `estimated_observations_per_month`. Enabled scanners sweep every 5 minutes, so a broad query can exhaust
-            the monthly quota quickly — run this and compare against `vision-quota-retrieve` before
-            `vision-scanners-create`.
+            an optional `sampling_rate` and `model`. Returns `matched_sessions_in_window`, the `window_days` it was
+            measured over, `estimated_observations_per_month`, `credits_per_observation` (the price at that model),
+            `estimated_credits_per_month`, and `other_enabled_scanners_monthly_credits`. Enabled scanners sweep every 5
+            minutes, so a broad query can exhaust the monthly budget quickly — run this and compare the *credit* figures
+            against `remaining` from `vision-quota-retrieve` before `vision-scanners-create`. Comparing observation
+            counts against `remaining` understates the cost: `remaining` is denominated in credits, and one observation
+            costs several.
         feature_flag: replay-vision
     vision-scanners-get:
         operation: vision_scanners_retrieve
@@ -398,8 +402,8 @@ tools:
             List all replay vision scanners in the current project. A scanner watches a slice of session recordings
             (defined by its `query` and `sampling_rate`) and runs one of four LLM analyses against each matching
             session: `monitor` (open-ended observations), `classifier` (tag from a fixed label set), `scorer` (numeric
-            rubric), or `summarizer` (free-text summary, optionally with facet embeddings for downstream search). Each
-            scanner returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.
+            rubric), or `summarizer` (free-text summary, with facet embeddings for downstream search). Each scanner
+            returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.
         feature_flag: replay-vision
     vision-scanners-observations-create-task-create:
         operation: vision_scanners_observations_create_task_create
diff --git a/products/replay_vision/skills/creating-replay-vision-scanners/SKILL.md b/products/replay_vision/skills/creating-replay-vision-scanners/SKILL.md
index 69f8446984c0..6aeaeb29a726 100644
--- a/products/replay_vision/skills/creating-replay-vision-scanners/SKILL.md
+++ b/products/replay_vision/skills/creating-replay-vision-scanners/SKILL.md
@@ -7,8 +7,9 @@ description: "Guides agents through creating and safely sizing a Replay Vision s
 
 A scanner is a standing LLM probe over session recordings. Once created and enabled, it runs on a
 **Temporal schedule that sweeps every 5 minutes**, applying its prompt to each new matching recording and
-recording the result as an observation (a queryable `$recording_observed` event). Each observation counts
-against a **monthly org quota** (a fixed number of observations per calendar month).
+recording the result as an observation (a queryable `$recording_observed` event). Each observation spends
+credits from a **monthly org credit budget** (1 credit = $0.01), and an observation's price depends on the
+scanner's model — so budget in credits, not in observation counts.
 
 That schedule is exactly why creation needs a gut-check: a scanner with a permissive query and full sampling
 starts consuming quota automatically and can drain the whole month's budget within its first few sweeps.
@@ -17,9 +18,9 @@ the budget may already be gone.
 
 ## Core principle: size before you ship
 
-Never create an enabled scanner blind. Estimate its volume, check remaining quota, and — when the projected
-volume is a meaningful fraction of what's left — show the user the numbers and get confirmation before
-creating. This is the heart of the skill; the rest is supporting detail.
+Never create an enabled scanner blind. Estimate its monthly credit spend, check the remaining credit budget,
+and — when the projected spend is a meaningful fraction of what's left — show the user the numbers and get
+confirmation before creating. This is the heart of the skill; the rest is supporting detail.
 
 ## The flow
 
@@ -32,11 +33,13 @@ Pick a `scanner_type` and write its `scanner_config`. Every type needs a `prompt
 | `monitor`    | Open-ended observation against a prompt (e.g. "flag rage clicks") | `{"prompt": "..."}`                                                                                                                     |
 | `classifier` | Assigns tags from a fixed label set                               | `{"prompt": "...", "tags": ["tag-a", "tag-b"]}` — `tags` needs ≥1 entry; optional `"multi_label": true`, `"allow_freeform_tags": false` |
 | `scorer`     | Numeric score on a rubric                                         | `{"prompt": "...", "scale": {"min": 1, "max": 5, "label": "frustration"}}` — `min` < `max`; `label` optional                            |
-| `summarizer` | Free-text summary; optional facet embeddings for search           | `{"prompt": "..."}`; optional `"length": "short" \| "medium" \| "long"` (default `"medium"`), `"emits_embeddings": false`               |
+| `summarizer` | Free-text summary plus facet embeddings for search                | `{"prompt": "..."}`; optional `"length": "short" \| "medium" \| "long"` (default `"medium"`)                                            |
+
+Summarizers always emit facet embeddings; there is no option to turn that off.
 
 `scanner_type` is **locked after creation** — to change it you delete and recreate, so confirm the type is
 right up front, and get the `scanner_config` shape right (a wrong shape is a create error, not a silent
-default).
+default — unknown keys are rejected too).
 
 If the user's intent makes the type and prompt obvious, just proceed — don't interrogate them.
 
@@ -54,19 +57,26 @@ trade coverage for budget.
 
 Before creating, run both checks and reason about them together:
 
-1. **Estimate volume** — call `vision-scanners-estimate-create` with the proposed `query` + `sampling_rate`.
-   It returns `matched_sessions_in_window`, the `window_days` measured, and
-   `estimated_observations_per_month`.
+1. **Estimate spend** — call `vision-scanners-estimate-create` with the proposed `query`, `sampling_rate`,
+   and `model`. It returns `matched_sessions_in_window`, the `window_days` measured,
+   `estimated_observations_per_month`, `credits_per_observation` (the price at that model), the resulting
+   `estimated_credits_per_month`, and `other_enabled_scanners_monthly_credits` (what the org's other enabled
+   scanners are already projected to spend).
 2. **Check budget** — call `vision-quota-retrieve` for `remaining` and `exhausted` against the org's monthly
    `credit_limit` (credits, 1 credit = $0.01; `null` when uncapped).
 
+Compare credits against credits — `remaining` is denominated in credits, not observations, so comparing it
+against `estimated_observations_per_month` understates the cost by the model's per-observation price.
+
 Then decide:
 
-- If `estimated_observations_per_month` comfortably fits within `remaining`, proceed.
+- If `estimated_credits_per_month` plus `other_enabled_scanners_monthly_credits` comfortably fits within
+  `remaining`, proceed.
 - If it's a large fraction of (or exceeds) `remaining`, **stop and tell the user the concrete numbers**
-  — e.g. "This scanner is projected to produce ~X observations/month; you have Y of Z left this month." —
-  and confirm before creating, or suggest tightening the `query` or lowering `sampling_rate` first.
-- If the org is already `exhausted`, say so — a new enabled scanner won't produce anything until the quota
+  — e.g. "This scanner is projected to spend ~X credits/month (~N observations at C credits each), on top of
+  ~Y credits from your other scanners; you have Z left this month." — and confirm before creating, or suggest
+  tightening the `query`, lowering `sampling_rate`, or picking a cheaper `model` first.
+- If the org is already `exhausted`, say so — a new enabled scanner won't produce anything until the budget
   resets, and its observations will be silently skipped.
 
 Confirmation here is a conversation step, not an API capability — surface the trade-off and let the user
diff --git a/services/mcp/schema/generated-tool-definitions.json b/services/mcp/schema/generated-tool-definitions.json
index 53bd3e37c22c..f4b1f09fb551 100644
--- a/services/mcp/schema/generated-tool-definitions.json
+++ b/services/mcp/schema/generated-tool-definitions.json
@@ -9752,7 +9752,7 @@
             "readOnlyHint": false
         },
         "requires_ai_consent": true,
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-delete": {
         "description": "Permanently delete a vision action by ID, along with its run history (past group summaries and alert firings). Its delivery destinations stop firing. To stop an action without losing its history, set `enabled: false` via `vision-actions-update` instead.",
@@ -9767,7 +9767,7 @@
             "openWorldHint": true,
             "readOnlyHint": false
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-list": {
         "description": "List the vision actions (the \"and then…\" automations that run over a scanner's observations) for the current project. A vision action in `mode: group_summary` produces the group summary shown on the \"Summaries and alerts\" tab — one report synthesized from up to `max_observations` observations of its bound scanner; `mode: alert` delivers only when its alert condition holds. Each row carries the bound `scanner`, `mode`, `trigger_type`, cadence, and `is_scanner_digest` (the built-in daily digest). Pass `?scanner=<id>` to filter to one scanner. To read a specific action's produced reports, list its runs with `vision-actions-runs-list`.",
@@ -9782,7 +9782,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-retrieve": {
         "description": "Get a single vision action by ID. Returns its configuration: the bound `scanner`, `mode` (group_summary/alert), `selection` (the observation-targeting predicate — verdict/tags/score bounds/window), `synthesis_config`, `alert_config`, cadence, and delivery targets. Use `vision-actions-runs-list` + `vision-actions-runs-retrieve` to read the reports it has actually produced.",
@@ -9797,7 +9797,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-runs-list": {
         "description": "List the run history for one vision action (nested under the action). Each run is one execution that produced a group summary or evaluated an alert. Rows are lightweight — `status` (running/completed/failed/skipped), `scheduled_at`, `observation_count` (how many observations fed the summary), `error_reason`, and `is_recovery` — without the report body. Fetch a completed run with `vision-actions-runs-retrieve` to get the synthesized report and the observations it cited.",
@@ -9812,7 +9812,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-runs-retrieve": {
         "description": "Get one vision action run in full: `synthesized_markdown` (the group-summary report, which cites its source observations inline with `[obs N]` markers) and `observations` — the recordings the run included, in summary order, each carrying the 1-based `index` that the report's `[obs N]` markers reference plus that observation's title and outcome. This is the tool for auditing a group summary: for each `[obs N]` in the markdown, `observations[N-1]` is the observation it cites, so you can check whether the cited observation actually supports the claim. Empty `synthesized_markdown`/`observations` means the run skipped, failed, or predates observation tracking.",
@@ -9827,7 +9827,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-update": {
         "description": "Partially update a vision action by ID. Send only the fields you want to change. Common changes: toggle `enabled` to pause/resume it, tune the `alert_config` threshold, change the `trigger_config` cadence, or point `delivery_config` at a different Slack channel or webhook. Nested objects and lists you send replace the stored value wholesale (e.g. `delivery_config` is the full new list), so fetch the action with `vision-actions-retrieve` first and resend the parts you want to keep. `mode` cross-field rules apply as on create: switching to `alert` requires an `alert_config`.",
@@ -9843,7 +9843,7 @@
             "readOnlyHint": false
         },
         "requires_ai_consent": true,
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-observations-label-create": {
         "description": "Set or update the shared thumbs up/down rating on a replay scanner observation: `is_correct` true means the scanner got the session right (thumbs up), false means it got it wrong (thumbs down), plus optional written `feedback` explaining why (applies to both directions). One shared rating per observation, team-wide, last write wins; re-posting updates it in place. These ratings power the scanner's calibration tab: the ratings-over-time chart and the AI prompt recommendations (see `vision-scanners-prompt-suggestions-generate`). Find observation IDs via `vision-scanners-observations-list` (filter `labeled=false` for unrated ones) or `vision-observations-list` for a session.",
@@ -9906,7 +9906,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-quota-retrieve": {
-        "description": "Get the organization's monthly Replay vision credit budget (1 credit = $0.01; observations cost credits by model). Returns `credit_limit` (null when uncapped), `credits_used` (in-flight + succeeded observations counted against it), `remaining`, `exhausted`, `projected_monthly_credits`, and the `period_start`/`period_end` of the current calendar-month window. Check `remaining`/`exhausted` before creating a scanner or calling `vision-scanners-scan-session` — observations requested over budget are silently skipped until the next period. Pair with `vision-scanners-estimate-create` to size a new scanner against the remaining budget.",
+        "description": "Get the organization's monthly Replay vision credit budget (1 credit = $0.01; observations cost credits by model). Returns `credit_limit` (null when uncapped), `credits_used` (in-flight + succeeded observations counted against it), `remaining`, `exhausted`, `projected_monthly_credits`, and the `period_start`/`period_end` of the current calendar-month window. Check `remaining`/`exhausted` before creating a scanner or calling `vision-scanners-scan-session`: over budget, a scanner's scheduled sweeps silently skip their observations until the next period, while `vision-scanners-scan-session` is rejected outright with a 402. Pair with `vision-scanners-estimate-create` to size a new scanner against the remaining budget.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Get replay vision quota",
@@ -9936,7 +9936,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-create": {
-        "description": "Create a replay vision scanner. Pick a `scanner_type` — `monitor` (open-ended observations), `classifier` (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary with optional facet embeddings via `emits_embeddings`) — and fill `scanner_config` to match: all types need a `prompt`; classifiers also need `tags`; scorers also need `scale`; summarizers optionally set `length` and `emits_embeddings`. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches — `date_from` and `date_to` are ignored (the schedule controls time). `sampling_rate` is a 0..1 random downsample applied after the query matches (default 1.0). `model` chooses the LLM. Names must be unique within the team. Before creating a broad scanner (a permissive `query` and/or high `sampling_rate`), first call `vision-scanners-estimate-create` to project its monthly observation volume and `vision-quota-retrieve` to check the org's remaining budget — an enabled scanner sweeps matching recordings every 5 minutes, and creation itself does not check quota, so a too-broad scanner can exhaust the monthly budget on its first sweeps. To get notified about what the scanner finds (a Slack alert or a scheduled summary), create a vision action bound to it with `vision-actions-create`.",
+        "description": "Create a replay vision scanner. Pick a `scanner_type` — `monitor` (open-ended observations), `classifier` (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, which always emits facet embeddings) — and fill `scanner_config` to match: all types need a `prompt`; classifiers also need `tags`; scorers also need `scale`; summarizers optionally set `length`. Unknown `scanner_config` keys are rejected. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches — `date_from` and `date_to` are ignored (the schedule controls time). `sampling_rate` is a 0..1 random downsample applied after the query matches (default 1.0). `model` chooses the LLM, and sets what each observation costs in credits. Names must be unique within the team. Before creating a broad scanner (a permissive `query` and/or high `sampling_rate`), first call `vision-scanners-estimate-create` to project its monthly credit spend and `vision-quota-retrieve` to check the org's remaining budget — an enabled scanner sweeps matching recordings every 5 minutes, and creation itself does not check quota, so a too-broad scanner can exhaust the monthly budget on its first sweeps. To get notified about what the scanner finds (a Slack alert or a scheduled summary), create a vision action bound to it with `vision-actions-create`.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Create replay vision scanner",
@@ -9967,7 +9967,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-estimate-create": {
-        "description": "Preview how many observations a proposed scanner would generate before saving it. Send a `query` (`RecordingsQuery` shape; `date_from`/`date_to` are ignored — the estimate uses a fixed 30-day lookback) and an optional `sampling_rate`. Returns `matched_sessions_in_window`, the `window_days` it was measured over, and `estimated_observations_per_month`. Enabled scanners sweep every 5 minutes, so a broad query can exhaust the monthly quota quickly — run this and compare against `vision-quota-retrieve` before `vision-scanners-create`.",
+        "description": "Preview how many observations a proposed scanner would generate before saving it. Send a `query` (`RecordingsQuery` shape; `date_from`/`date_to` are ignored — the estimate uses a fixed 30-day lookback) and an optional `sampling_rate` and `model`. Returns `matched_sessions_in_window`, the `window_days` it was measured over, `estimated_observations_per_month`, `credits_per_observation` (the price at that model), `estimated_credits_per_month`, and `other_enabled_scanners_monthly_credits`. Enabled scanners sweep every 5 minutes, so a broad query can exhaust the monthly budget quickly — run this and compare the *credit* figures against `remaining` from `vision-quota-retrieve` before `vision-scanners-create`. Comparing observation counts against `remaining` understates the cost: `remaining` is denominated in credits, and one observation costs several.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Estimate replay vision scanner volume",
@@ -10012,7 +10012,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-list": {
-        "description": "List all replay vision scanners in the current project. A scanner watches a slice of session recordings (defined by its `query` and `sampling_rate`) and runs one of four LLM analyses against each matching session: `monitor` (open-ended observations), `classifier` (tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, optionally with facet embeddings for downstream search). Each scanner returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.",
+        "description": "List all replay vision scanners in the current project. A scanner watches a slice of session recordings (defined by its `query` and `sampling_rate`) and runs one of four LLM analyses against each matching session: `monitor` (open-ended observations), `classifier` (tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, with facet embeddings for downstream search). Each scanner returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "List replay vision scanners",
diff --git a/services/mcp/schema/tool-definitions-all.json b/services/mcp/schema/tool-definitions-all.json
index e0b3fa3f416b..91b16a6fe878 100644
--- a/services/mcp/schema/tool-definitions-all.json
+++ b/services/mcp/schema/tool-definitions-all.json
@@ -10392,7 +10392,7 @@
             "readOnlyHint": false
         },
         "requires_ai_consent": true,
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-delete": {
         "description": "Permanently delete a vision action by ID, along with its run history (past group summaries and alert firings). Its delivery destinations stop firing. To stop an action without losing its history, set `enabled: false` via `vision-actions-update` instead.",
@@ -10407,7 +10407,7 @@
             "openWorldHint": true,
             "readOnlyHint": false
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-list": {
         "description": "List the vision actions (the \"and then…\" automations that run over a scanner's observations) for the current project. A vision action in `mode: group_summary` produces the group summary shown on the \"Summaries and alerts\" tab — one report synthesized from up to `max_observations` observations of its bound scanner; `mode: alert` delivers only when its alert condition holds. Each row carries the bound `scanner`, `mode`, `trigger_type`, cadence, and `is_scanner_digest` (the built-in daily digest). Pass `?scanner=<id>` to filter to one scanner. To read a specific action's produced reports, list its runs with `vision-actions-runs-list`.",
@@ -10422,7 +10422,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-retrieve": {
         "description": "Get a single vision action by ID. Returns its configuration: the bound `scanner`, `mode` (group_summary/alert), `selection` (the observation-targeting predicate — verdict/tags/score bounds/window), `synthesis_config`, `alert_config`, cadence, and delivery targets. Use `vision-actions-runs-list` + `vision-actions-runs-retrieve` to read the reports it has actually produced.",
@@ -10437,7 +10437,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-runs-list": {
         "description": "List the run history for one vision action (nested under the action). Each run is one execution that produced a group summary or evaluated an alert. Rows are lightweight — `status` (running/completed/failed/skipped), `scheduled_at`, `observation_count` (how many observations fed the summary), `error_reason`, and `is_recovery` — without the report body. Fetch a completed run with `vision-actions-runs-retrieve` to get the synthesized report and the observations it cited.",
@@ -10452,7 +10452,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-runs-retrieve": {
         "description": "Get one vision action run in full: `synthesized_markdown` (the group-summary report, which cites its source observations inline with `[obs N]` markers) and `observations` — the recordings the run included, in summary order, each carrying the 1-based `index` that the report's `[obs N]` markers reference plus that observation's title and outcome. This is the tool for auditing a group summary: for each `[obs N]` in the markdown, `observations[N-1]` is the observation it cites, so you can check whether the cited observation actually supports the claim. Empty `synthesized_markdown`/`observations` means the run skipped, failed, or predates observation tracking.",
@@ -10467,7 +10467,7 @@
             "openWorldHint": true,
             "readOnlyHint": true
         },
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-actions-update": {
         "description": "Partially update a vision action by ID. Send only the fields you want to change. Common changes: toggle `enabled` to pause/resume it, tune the `alert_config` threshold, change the `trigger_config` cadence, or point `delivery_config` at a different Slack channel or webhook. Nested objects and lists you send replace the stored value wholesale (e.g. `delivery_config` is the full new list), so fetch the action with `vision-actions-retrieve` first and resend the parts you want to keep. `mode` cross-field rules apply as on create: switching to `alert` requires an `alert_config`.",
@@ -10483,7 +10483,7 @@
             "readOnlyHint": false
         },
         "requires_ai_consent": true,
-        "feature_flag": "replay-vision"
+        "feature_flag": "replay-vision-actions"
     },
     "vision-observations-label-create": {
         "description": "Set or update the shared thumbs up/down rating on a replay scanner observation: `is_correct` true means the scanner got the session right (thumbs up), false means it got it wrong (thumbs down), plus optional written `feedback` explaining why (applies to both directions). One shared rating per observation, team-wide, last write wins; re-posting updates it in place. These ratings power the scanner's calibration tab: the ratings-over-time chart and the AI prompt recommendations (see `vision-scanners-prompt-suggestions-generate`). Find observation IDs via `vision-scanners-observations-list` (filter `labeled=false` for unrated ones) or `vision-observations-list` for a session.",
@@ -10546,7 +10546,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-quota-retrieve": {
-        "description": "Get the organization's monthly Replay vision credit budget (1 credit = $0.01; observations cost credits by model). Returns `credit_limit` (null when uncapped), `credits_used` (in-flight + succeeded observations counted against it), `remaining`, `exhausted`, `projected_monthly_credits`, and the `period_start`/`period_end` of the current calendar-month window. Check `remaining`/`exhausted` before creating a scanner or calling `vision-scanners-scan-session` — observations requested over budget are silently skipped until the next period. Pair with `vision-scanners-estimate-create` to size a new scanner against the remaining budget.",
+        "description": "Get the organization's monthly Replay vision credit budget (1 credit = $0.01; observations cost credits by model). Returns `credit_limit` (null when uncapped), `credits_used` (in-flight + succeeded observations counted against it), `remaining`, `exhausted`, `projected_monthly_credits`, and the `period_start`/`period_end` of the current calendar-month window. Check `remaining`/`exhausted` before creating a scanner or calling `vision-scanners-scan-session`: over budget, a scanner's scheduled sweeps silently skip their observations until the next period, while `vision-scanners-scan-session` is rejected outright with a 402. Pair with `vision-scanners-estimate-create` to size a new scanner against the remaining budget.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Get replay vision quota",
@@ -10576,7 +10576,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-create": {
-        "description": "Create a replay vision scanner. Pick a `scanner_type` — `monitor` (open-ended observations), `classifier` (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary with optional facet embeddings via `emits_embeddings`) — and fill `scanner_config` to match: all types need a `prompt`; classifiers also need `tags`; scorers also need `scale`; summarizers optionally set `length` and `emits_embeddings`. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches — `date_from` and `date_to` are ignored (the schedule controls time). `sampling_rate` is a 0..1 random downsample applied after the query matches (default 1.0). `model` chooses the LLM. Names must be unique within the team. Before creating a broad scanner (a permissive `query` and/or high `sampling_rate`), first call `vision-scanners-estimate-create` to project its monthly observation volume and `vision-quota-retrieve` to check the org's remaining budget — an enabled scanner sweeps matching recordings every 5 minutes, and creation itself does not check quota, so a too-broad scanner can exhaust the monthly budget on its first sweeps. To get notified about what the scanner finds (a Slack alert or a scheduled summary), create a vision action bound to it with `vision-actions-create`.",
+        "description": "Create a replay vision scanner. Pick a `scanner_type` — `monitor` (open-ended observations), `classifier` (assign a tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, which always emits facet embeddings) — and fill `scanner_config` to match: all types need a `prompt`; classifiers also need `tags`; scorers also need `scale`; summarizers optionally set `length`. Unknown `scanner_config` keys are rejected. `query` is a `RecordingsQuery` shape that defines which sessions the scanner watches — `date_from` and `date_to` are ignored (the schedule controls time). `sampling_rate` is a 0..1 random downsample applied after the query matches (default 1.0). `model` chooses the LLM, and sets what each observation costs in credits. Names must be unique within the team. Before creating a broad scanner (a permissive `query` and/or high `sampling_rate`), first call `vision-scanners-estimate-create` to project its monthly credit spend and `vision-quota-retrieve` to check the org's remaining budget — an enabled scanner sweeps matching recordings every 5 minutes, and creation itself does not check quota, so a too-broad scanner can exhaust the monthly budget on its first sweeps. To get notified about what the scanner finds (a Slack alert or a scheduled summary), create a vision action bound to it with `vision-actions-create`.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Create replay vision scanner",
@@ -10607,7 +10607,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-estimate-create": {
-        "description": "Preview how many observations a proposed scanner would generate before saving it. Send a `query` (`RecordingsQuery` shape; `date_from`/`date_to` are ignored — the estimate uses a fixed 30-day lookback) and an optional `sampling_rate`. Returns `matched_sessions_in_window`, the `window_days` it was measured over, and `estimated_observations_per_month`. Enabled scanners sweep every 5 minutes, so a broad query can exhaust the monthly quota quickly — run this and compare against `vision-quota-retrieve` before `vision-scanners-create`.",
+        "description": "Preview how many observations a proposed scanner would generate before saving it. Send a `query` (`RecordingsQuery` shape; `date_from`/`date_to` are ignored — the estimate uses a fixed 30-day lookback) and an optional `sampling_rate` and `model`. Returns `matched_sessions_in_window`, the `window_days` it was measured over, `estimated_observations_per_month`, `credits_per_observation` (the price at that model), `estimated_credits_per_month`, and `other_enabled_scanners_monthly_credits`. Enabled scanners sweep every 5 minutes, so a broad query can exhaust the monthly budget quickly — run this and compare the *credit* figures against `remaining` from `vision-quota-retrieve` before `vision-scanners-create`. Comparing observation counts against `remaining` understates the cost: `remaining` is denominated in credits, and one observation costs several.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "Estimate replay vision scanner volume",
@@ -10652,7 +10652,7 @@
         "feature_flag": "replay-vision"
     },
     "vision-scanners-list": {
-        "description": "List all replay vision scanners in the current project. A scanner watches a slice of session recordings (defined by its `query` and `sampling_rate`) and runs one of four LLM analyses against each matching session: `monitor` (open-ended observations), `classifier` (tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, optionally with facet embeddings for downstream search). Each scanner returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.",
+        "description": "List all replay vision scanners in the current project. A scanner watches a slice of session recordings (defined by its `query` and `sampling_rate`) and runs one of four LLM analyses against each matching session: `monitor` (open-ended observations), `classifier` (tag from a fixed label set), `scorer` (numeric rubric), or `summarizer` (free-text summary, with facet embeddings for downstream search). Each scanner returns its current configuration, enabled state, sampling rate, and `last_swept_at` timestamp.",
         "category": "Replay vision",
         "feature": "replay_vision",
         "summary": "List replay vision scanners",
diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts
index 718c9cdbe23a..ba0096144c80 100644
--- a/services/mcp/src/api/generated.ts
+++ b/services/mcp/src/api/generated.ts
@@ -61011,6 +61011,14 @@ export namespace Schemas {
       layout?: LayoutEnum;
     }
 
+    /**
+     * The shape every Replay Vision error response uses, so generated clients read one key.
+     */
+    export interface ReplayVisionError {
+      /** Human-readable explanation of why the request was refused. */
+      detail: string;
+    }
+
     export interface ReplayVisionScannerFindingSignalExtra {
       scanner_id: string;
       scanner_name: string;
diff --git a/services/mcp/tests/unit/tool-filtering.test.ts b/services/mcp/tests/unit/tool-filtering.test.ts
index 7c9d83b977d1..08450ff1fa23 100644
--- a/services/mcp/tests/unit/tool-filtering.test.ts
+++ b/services/mcp/tests/unit/tool-filtering.test.ts
@@ -840,6 +840,7 @@ describe('Tool Filtering - Feature Flags', () => {
                 'notebooks-collaboration',
                 'revamped-py-notebooks',
                 'replay-vision',
+                'replay-vision-actions',
                 'tasks',
                 'dashboard-widgets',
                 'heatmaps-mcp',
@@ -859,7 +860,7 @@ describe('Tool Filtering - Feature Flags', () => {
                 'streamlit-apps',
             ])
         )
-        expect(flags).toHaveLength(26)
+        expect(flags).toHaveLength(27)
     })
 
     it('every loops tool is gated on the loops flag', () => {