insights/products/signals/backend/views.py

import json
import uuid
from collections.abc import Callable, Sequence
from datetime import UTC, timedelta
from typing import Any, cast

from django.conf import settings
from django.core.cache import cache
from django.db import IntegrityError, models, transaction
from django.db.models import (
    BooleanField,
    Case,
    CharField,
    Count,
    Exists,
    F,
    Func,
    IntegerField,
    JSONField,
    OuterRef,
    Prefetch,
    Q,
    Subquery,
    Sum,
    Value,
    When,
)
from django.db.models.functions import Cast, Coalesce
from django.utils import timezone

import structlog
import hanzo_insights
from asgiref.sync import async_to_sync
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema, extend_schema_view
from opentelemetry import trace
from rest_framework import exceptions, mixins, serializers, status, viewsets
from rest_framework.authentication import SessionAuthentication
from rest_framework.decorators import action
from rest_framework.exceptions import NotFound
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView
from temporalio.common import RetryPolicy, WorkflowIDConflictPolicy, WorkflowIDReusePolicy
from temporalio.exceptions import WorkflowAlreadyStartedError

from insights.api.integration import github_rate_limited_response
from insights.api.mixins import ValidatedRequest, validated_request
from insights.api.routing import TeamAndOrgViewSetMixin
from insights.auth import InternalAPIAuthentication, OAuthAccessTokenAuthentication, PersonalAPIKeyAuthentication
from insights.egress.github.transport import GitHubEgressBudgetExhausted, GitHubRateLimitError
from insights.egress.limiter.policies import Priority
from insights.event_usage import report_user_action
from insights.exceptions_capture import capture_exception
from insights.models import Team, User
from insights.models.activity_logging.activity_log import Change, Detail, log_activity
from insights.models.activity_logging.model_activity import is_impersonated_session
from insights.models.github_integration_base import GitHubIntegrationBase
from insights.models.integration import GitHubIntegration, Integration
from insights.models.team.extensions import get_or_create_team_extension
from insights.models.user_integration import ReauthorizationRequired, UserGitHubIntegration, UserIntegration
from insights.permissions import APIScopePermission, get_authenticator_scoped_team_ids, get_authenticator_scopes
from insights.temporal.common.client import sync_connect
from insights.user_permissions import UserPermissions

from products.data_warehouse.backend.facade.api import trigger_external_data_workflow
from products.signals.backend.artefact_schemas import (
    NON_WRITABLE_ARTEFACT_TYPES,
    ArtefactContentValidationError,
    Dismissal,
    SuggestedReviewers,
    SummaryChange,
    TitleChange,
    parse_artefact_content,
)
from products.signals.backend.billing import (
    REFUND_INELIGIBLE_BILLING_EXEMPT,
    REFUND_INELIGIBLE_NO_BILLABLE_PR,
    REFUND_INELIGIBLE_OUT_OF_PERIOD,
    SIGNALS_CREDITS_PER_REPORT_WITH_PR,
    annotate_first_billable_pr_run_at,
    current_billing_period_bounds,
    first_billable_pr_run,
    period_billable_credits_for_org,
    refund_ineligibility_reason,
    report_pr_is_merged,
)
from products.signals.backend.dismissal_notes import forward_dismissal_note
from products.signals.backend.facade.api import emit_signal
from products.signals.backend.feedback_notes import forward_feedback_note
from products.signals.backend.implementation_pr import (
    fetch_implementation_pr_state_for_reports,
    fetch_implementation_pr_urls_for_reports,
)
from products.signals.backend.models import (
    ArtefactAttribution,
    AutonomyPriority,
    InvalidStatusTransition,
    SignalReport,
    SignalReportArtefact,
    SignalReportRefund,
    SignalSourceConfig,
    SignalTeamConfig,
    SignalUserAutonomyConfig,
)
from products.signals.backend.quota import self_driving_quota_enforcement_enabled, self_driving_quota_gate
from products.signals.backend.report_generation.research import ActionabilityChoice
from products.signals.backend.report_generation.resolve_reviewers import (
    get_org_member_github_login_to_user_map,
    get_org_member_github_logins_by_user_uuid,
    normalized_github_logins_from_suggested_reviewer_artefacts,
    resolve_org_github_login_to_users,
)
from products.signals.backend.serializers import (
    CommitDiffResponseSerializer,
    PullRequestChecksResponseSerializer,
    PullRequestCommentsResponseSerializer,
    PullRequestReviewCommentCreateResponseSerializer,
    PullRequestReviewCommentCreateSerializer,
    PullRequestReviewCommentReactionCreateResponseSerializer,
    PullRequestReviewCommentReactionCreateSerializer,
    PullRequestReviewCommentUpdateSerializer,
    ReportSignalsResponseSerializer,
    SignalReportArtefactLogCreateSerializer,
    SignalReportArtefactLogUpdateSerializer,
    SignalReportArtefactSerializer,
    SignalReportArtefactWriteResponseSerializer,
    SignalReportArtefactWriteSerializer,
    SignalReportRefundSerializer,
    SignalReportSerializer,
    SignalSourceConfigSerializer,
    SignalTeamConfigSerializer,
    SignalUserAutonomyConfigCreateSerializer,
    SignalUserAutonomyConfigSerializer,
)
from products.signals.backend.signal_metadata import fetch_source_products_for_reports
from products.signals.backend.task_attribution import (
    TASK_ID_HEADER,
    resolve_request_attribution,
    resolve_task_id_from_header,
)
from products.signals.backend.tasks import send_reviewer_added_slack_notifications, sync_signals_refund_credit
from products.signals.backend.temporal.backfill_error_tracking import (
    BackfillErrorTrackingInput,
    BackfillErrorTrackingWorkflow,
)
from products.signals.backend.temporal.deletion import SignalReportDeletionWorkflow
from products.signals.backend.temporal.grouping_v2 import TeamSignalGroupingV2Workflow
from products.signals.backend.temporal.reingestion import SignalReportReingestionWorkflow
from products.signals.backend.temporal.signal_queries import (
    fetch_live_report_ids_for_source_ids,
    fetch_report_ids_for_scout_names,
    fetch_report_ids_for_scout_prefix,
    fetch_report_ids_for_source_products,
    fetch_signals_for_report_sync,
)
from products.signals.backend.temporal.types import (
    SignalReportDeletionWorkflowInputs,
    SignalReportReingestionWorkflowInputs,
)
from products.tasks.backend.facade import api as tasks_facade
from products.warehouse_sources.backend.facade.models import ExternalDataSchema

logger = structlog.get_logger(__name__)
tracer = trace.get_tracer(__name__)

# `available_reviewers` returns every eligible org member in a single unpaginated payload.
# Org membership is tiny in practice, so even the biggest
# org today serialises to well under 100 KB. If an org ever exceeds this threshold we want a
# signal that it's time to add real pagination, rather than silently truncating the list (the
# old behaviour, which capped at 100 and dropped everyone alphabetically after ~"M").
REVIEWER_PAGINATION_THRESHOLD = 1200
PR_GITHUB_CACHE_SECONDS = 15


class EmitSignalSerializer(serializers.Serializer):
    source_product = serializers.CharField(max_length=100)
    source_type = serializers.CharField(max_length=100)
    description = serializers.CharField()
    weight = serializers.FloatField(default=0.5, min_value=0.0, max_value=1.0)
    extra = serializers.DictField(required=False, default=dict)


# Simple debug view, to make testing out the flow easier. Disabled in production.
class SignalViewSet(TeamAndOrgViewSetMixin, viewsets.ViewSet):
    scope_object = "INTERNAL"

    @extend_schema(exclude=True)
    @action(methods=["POST"], detail=False)
    def emit(self, request: Request, *args, **kwargs):
        if not settings.DEBUG:
            raise NotFound()

        serializer = EmitSignalSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        data = serializer.validated_data

        async_to_sync(emit_signal)(
            team=self.team,
            source_product=data["source_product"],
            source_type=data["source_type"],
            source_id=str(uuid.uuid4()),
            description=data["description"],
            weight=data["weight"],
            extra=data["extra"],
        )

        return Response({"status": "ok"}, status=status.HTTP_202_ACCEPTED)


class InternalEmitSignalSerializer(serializers.Serializer):
    source_product = serializers.CharField(max_length=100)
    source_type = serializers.CharField(max_length=100)
    source_id = serializers.CharField(max_length=512)
    description = serializers.CharField()
    weight = serializers.FloatField(default=0.5, min_value=0.0, max_value=1.0)
    extra = serializers.DictField(required=False, default=dict)


class InternalSignalViewSet(viewsets.ViewSet):
    """
    Internal-only endpoint for service-to-service signal emission (e.g. from cymbal).
    Authenticated via X-Internal-Api-Secret header, not exposed to external ingress.
    """

    authentication_classes = [InternalAPIAuthentication]

    @extend_schema(exclude=True)
    def emit(self, request: Request, team_id: str, *args, **kwargs):
        try:
            team = Team.objects.get(id=int(team_id))
        except (Team.DoesNotExist, ValueError):
            return Response({"error": "Team not found"}, status=status.HTTP_404_NOT_FOUND)

        serializer = InternalEmitSignalSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        data = serializer.validated_data

        async_to_sync(emit_signal)(
            team=team,
            source_product=data["source_product"],
            source_type=data["source_type"],
            source_id=data["source_id"],
            description=data["description"],
            weight=data["weight"],
            extra=data["extra"],
        )

        return Response({"status": "ok"}, status=status.HTTP_202_ACCEPTED)


class SignalSourceConfigViewSet(TeamAndOrgViewSetMixin, viewsets.ModelViewSet):
    serializer_class = SignalSourceConfigSerializer
    authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication]
    permission_classes = [IsAuthenticated, APIScopePermission]
    scope_object = "task"
    queryset = SignalSourceConfig.objects.all().order_by("-updated_at")

    @tracer.start_as_current_span("signals.source_configs.list")
    def list(self, request, *args, **kwargs):
        # This list is fetched on inbox load. The default serializer resolves a per-row `status`,
        # which for session-analysis rows makes a synchronous Temporal RPC — a potential N+1. The
        # span lets us see how much of the inbox load this endpoint accounts for.
        return super().list(request, *args, **kwargs)

    def _is_scout_source(self, source_product: str | None, source_type: str | None) -> bool:
        return (
            source_product == SignalSourceConfig.SourceProduct.SIGNALS_SCOUT
            and source_type == SignalSourceConfig.SourceType.CROSS_SOURCE_ISSUE
        )

    def _config_team_id(self, source_product: str | None, source_type: str | None) -> int:
        # The scout source config is a project-level singleton: the scout fleet canonicalizes
        # child environments to the parent team, and the emit preflight gates on the parent
        # team's row (see scout_harness/views.py `_canonical_team_id` and tools/emit.py). Writing
        # it to the canonical team keeps the inbox toggle and the emit gate on the same row from
        # any environment. All other sources stay environment-scoped.
        if self._is_scout_source(source_product, source_type):
            return self.team.parent_team_id or self.team_id
        return self.team_id

    def _filter_queryset_by_parents_lookups(self, queryset):
        # Mirror of `_config_team_id` on the read side: surface the scout row from the canonical
        # (parent) team while every other source stays scoped to the URL environment, so the
        # toggle reads and updates the same project-level row the emit gate checks.
        canonical_team_id = self.team.parent_team_id or self.team_id
        scout_source = Q(
            source_product=SignalSourceConfig.SourceProduct.SIGNALS_SCOUT,
            source_type=SignalSourceConfig.SourceType.CROSS_SOURCE_ISSUE,
        )
        return queryset.filter(
            (Q(team_id=self.team_id) & ~scout_source) | (Q(team_id=canonical_team_id) & scout_source)
        )

    def perform_create(self, serializer):
        team_id = self._config_team_id(
            serializer.validated_data.get("source_product"), serializer.validated_data.get("source_type")
        )
        try:
            instance = serializer.save(team_id=team_id, created_by=self.request.user)
        except IntegrityError:
            raise serializers.ValidationError(
                {"source_product": "A configuration for this source product and type already exists for this team."}
            )

        if (
            instance.source_product == SignalSourceConfig.SourceProduct.ERROR_TRACKING
            and instance.source_type == SignalSourceConfig.SourceType.ISSUE_CREATED
            and instance.enabled
        ):
            self._trigger_error_tracking_backfill()

    def _trigger_error_tracking_backfill(self) -> None:
        """Fire-and-forget backfill of recent error tracking issues as signals."""
        try:
            client = sync_connect()
            async_to_sync(client.start_workflow)(  # type: ignore
                "backfill-error-tracking",  # type: ignore
                BackfillErrorTrackingInput(team_id=self.team_id),  # type: ignore
                id=BackfillErrorTrackingWorkflow.workflow_id_for(self.team_id),
                id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
                id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
                task_queue=settings.VIDEO_EXPORT_TASK_QUEUE,
                retry_policy=RetryPolicy(maximum_attempts=1),
            )
            logger.info(f"Started error tracking backfill workflow for team {self.team_id}")
        except Exception:
            logger.exception(f"Failed to start error tracking backfill workflow for team {self.team_id}")

    def perform_update(self, serializer):
        instance = cast(SignalSourceConfig, serializer.instance)
        was_enabled = instance.enabled

        # The source keys are the row's identity and decide its canonical team_id on create
        # (`_config_team_id`); retagging them in place would strand the row on the wrong team —
        # e.g. a child-environment row retagged to the scout source would stay on the child team,
        # hidden by the read filter while the emit gate checks the parent. There is no legitimate
        # reason to change a config's source identity, so reject it rather than re-deriving team_id.
        for field in ("source_product", "source_type"):
            new_value = serializer.validated_data.get(field)
            if new_value is not None and new_value != getattr(instance, field):
                raise serializers.ValidationError({field: f"{field} cannot be changed after creation."})

        try:
            instance = serializer.save()
        except IntegrityError:
            raise serializers.ValidationError(
                {"source_product": "A configuration for this source product and type already exists for this team."}
            )

        # SESSION_ANALYSIS_CLUSTER used to upsert a summarization schedule here. That
        # sweep ran the enterprise summarization pipeline, which is not in this fork, so
        # the branch had nothing left to schedule. Data-import sources are unaffected.
        if (
            instance.enabled
            and not was_enabled
            and instance.source_type != SignalSourceConfig.SourceType.SESSION_ANALYSIS_CLUSTER
        ):
            self._trigger_data_import_sync(instance)

    # Maps source_product to ExternalDataSourceType value for data import sources
    _DATA_IMPORT_SOURCE_TYPE_MAP: dict[str, str] = {
        SignalSourceConfig.SourceProduct.GITHUB: "Github",
        SignalSourceConfig.SourceProduct.LINEAR: "Linear",
        SignalSourceConfig.SourceProduct.ZENDESK: "Zendesk",
    }

    def _trigger_data_import_sync(self, config: SignalSourceConfig) -> None:
        """Fire-and-forget sync trigger for data import signal sources."""
        ext_source_type = self._DATA_IMPORT_SOURCE_TYPE_MAP.get(config.source_product)
        if ext_source_type is None:
            return

        schemas = (
            ExternalDataSchema.objects.filter(
                team_id=self.team_id,
                source__source_type=ext_source_type,
                should_sync=True,
            )
            .exclude(source__deleted=True)
            .select_related("source")
        )
        for schema in schemas:
            try:
                trigger_external_data_workflow(schema)
                logger.info("Triggered data import sync for %s schema %s", config.source_product, schema.id)
            except Exception:
                logger.exception(
                    "Failed to trigger data import sync for %s schema %s", config.source_product, schema.id
                )


class SignalTeamConfigViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet):
    """Team-level signal autonomy config (singleton per team).

    GET  /signals/config/  → retrieve
    POST /signals/config/  → update
    """

    authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication]
    permission_classes = [IsAuthenticated, APIScopePermission]
    serializer_class = SignalTeamConfigSerializer
    queryset = SignalTeamConfig.objects.all()
    scope_object = "task"

    def dangerously_get_required_scopes(self, request: Request, view) -> list[str] | None:
        if request.method in ("GET", "HEAD", "OPTIONS"):
            return ["task:read"]
        return ["task:write"]

    def _get_config(self) -> SignalTeamConfig:
        # Singleton per team with safe defaults. A post_save signal creates it on team
        # creation, but teams predating that signal (or where it failed) have no row, so
        # lazily create it here — otherwise the first read/write (e.g. connecting a default
        # notification channel) would 404.
        return get_or_create_team_extension(self.team, SignalTeamConfig)

    @extend_schema(exclude=True)
    def list(self, request: Request, *args, **kwargs) -> Response:
        return Response(SignalTeamConfigSerializer(self._get_config()).data)

    @extend_schema(exclude=True)
    def create(self, request: Request, *args, **kwargs) -> Response:
        config = self._get_config()
        serializer = SignalTeamConfigSerializer(config, data=request.data, partial=True)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data)


SIGNAL_REPORT_DISMISSAL_NOTE_MAX_LENGTH = 4000
# Upper bound on how far a snooze can push out re-promotion. Generous enough for any
# realistic snooze, but bounded so a caller can't effectively block a report forever.
SIGNAL_REPORT_MAX_SNOOZE_FOR = 100_000
# Upper bound on how many reports a single bulk transition may touch. Keeps one call
# from fanning out into an unbounded write; callers page through larger sets.
SIGNAL_REPORT_BULK_STATE_MAX_IDS = 100
# Bounds on the editable human-facing report fields. `title`/`summary` are TextFields on the
# model (so unbounded in the DB), but the write API caps them to keep an edit from storing an
# absurdly long title or summary.
SIGNAL_REPORT_TITLE_MAX_LENGTH = 300
SIGNAL_REPORT_SUMMARY_MAX_LENGTH = 10_000

# Canonical dismissal reason codes, mirrored from the inbox UI source of truth at
# frontend/src/scenes/inbox/utils/dismissalReasons.ts (itself a port of desktop's
# packages/shared/src/dismissal-reasons.ts). Constraining the API to these values keeps
# agent-supplied reasons rendering as labelled chips in the inbox instead of raw,
# unrecognised codes. Keep the values (and order) in sync with that file.
SIGNAL_REPORT_DISMISSAL_REASON_CHOICES = [
    ("already_fixed", "Already fixed"),
    ("report_unclear", "Report is unclear to me"),
    ("analysis_wrong", "Agent's analysis is wrong"),
    ("wontfix_intentional", "Won't fix - intentional behavior"),
    ("wontfix_irrelevant", "Won't fix - issue is real but insignificant"),
    ("other", "Something else…"),
]

_DISMISSAL_REASON_HELP_TEXT = (
    "Optional canonical reason code for the dismissal. Must be one of: already_fixed, "
    "report_unclear, analysis_wrong, wontfix_intentional, wontfix_irrelevant, other — these match "
    "the inbox UI so the rationale renders as a labelled chip rather than a raw code. When the work "
    "this report asked for is done, the honest transition is state='resolved' (the reason/note records "
    "why). Reserve 'already_fixed' with state='potential' (snooze/restore) for \"fixed by something "
    "else / might recur\" cases, so the report reappears if the issue comes back. Use 'other' together "
    "with a dismissal_note for anything that doesn't fit a code."
)


class SignalReportBulkStateOutcome(models.TextChoices):
    """Per-id result of a bulk state transition. Mirrors the single-report responses:
    `transitioned` ~ 200, `skipped` ~ 409 (transition not allowed), `failed` ~ 400."""

    TRANSITIONED = "transitioned", "transitioned"
    SKIPPED = "skipped", "skipped"
    FAILED = "failed", "failed"
    NOT_FOUND = "not_found", "not_found"


# Statuses a report may have held before being archived and still resolve straight out of the
# archive. Mirrors the model's direct resolve edge (pending_input | ready -> resolved) plus RESOLVED
# itself, which makes archive-then-resolve idempotent. FAILED is deliberately absent: the model
# refuses FAILED -> RESOLVED directly, and suppression must not launder a failed report into looking
# successfully resolved.
_RESOLVABLE_STATUSES_BEFORE_SUPPRESSION = frozenset(
    {
        SignalReport.Status.READY,
        SignalReport.Status.PENDING_INPUT,
        SignalReport.Status.RESOLVED,
    }
)


class SignalReportStateRequestSerializer(serializers.Serializer):
    state = serializers.ChoiceField(
        choices=[("suppressed", "suppressed"), ("potential", "potential"), ("resolved", "resolved")],
        help_text=(
            "Target state for the report. Use 'suppressed' to dismiss the report from the inbox, "
            "'potential' to snooze/reopen it for later review, or 'resolved' when the work this report "
            "asked for has been done. Resolving is only allowed from a researched status (ready or "
            "pending_input) or a suppressed report; other statuses return 409 (skipped in bulk)."
        ),
    )
    dismissal_reason = serializers.ChoiceField(
        required=False,
        choices=SIGNAL_REPORT_DISMISSAL_REASON_CHOICES,
        help_text=_DISMISSAL_REASON_HELP_TEXT,
    )
    dismissal_note = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=SIGNAL_REPORT_DISMISSAL_NOTE_MAX_LENGTH,
        help_text="Optional free-form note explaining the dismissal. Capped at 4000 characters.",
    )
    snooze_for = serializers.IntegerField(
        required=False,
        min_value=1,
        max_value=SIGNAL_REPORT_MAX_SNOOZE_FOR,
        help_text=(
            "Optional, only honored when state is 'potential'. Number of additional signals the report "
            "must accumulate before it is re-promoted into the pipeline — effectively snoozing it until then. "
            "Omit to let the report re-enter the pipeline on the next matching signal."
        ),
    )


class SignalReportBulkStateRequestSerializer(SignalReportStateRequestSerializer):
    ids = serializers.ListField(
        child=serializers.UUIDField(),
        allow_empty=False,
        max_length=SIGNAL_REPORT_BULK_STATE_MAX_IDS,
        help_text=(
            "Report ids to transition to `state` in one call (1–"
            f"{SIGNAL_REPORT_BULK_STATE_MAX_IDS}). Duplicates are de-duplicated; each id is "
            "processed independently so one disallowed transition does not block the rest. "
            "`dismissal_reason`, `dismissal_note` and `snooze_for` apply to every id."
        ),
    )


class SignalReportBulkStateResultSerializer(serializers.Serializer):
    id = serializers.UUIDField(help_text="The report id this result refers to.")
    # CharField (not ChoiceField) on purpose: this is a server-generated response value, so it
    # needs no input validation, and a ChoiceField named `outcome` would collide with another
    # product's `OutcomeEnum` in the shared OpenAPI schema. Values come from SignalReportBulkStateOutcome.
    outcome = serializers.CharField(
        help_text=(
            "One of: transitioned, skipped, failed, not_found. transitioned: the state change was applied. "
            "skipped: the transition was not allowed from the report's current status (a 409 on the "
            "single-report endpoint). failed: the request data was invalid for this report. not_found: no "
            "report with this id is visible to you."
        ),
    )
    status = serializers.CharField(
        required=False,
        allow_null=True,
        help_text="The report's status after the transition. Present only when outcome is 'transitioned'.",
    )
    detail = serializers.CharField(
        required=False,
        allow_null=True,
        help_text="Human-readable explanation for non-transitioned outcomes (skipped / failed / not_found).",
    )


class SignalReportBulkStateResponseSerializer(serializers.Serializer):
    results = SignalReportBulkStateResultSerializer(
        many=True,
        help_text="One result per requested id, in request order (after de-duplication).",
    )
    transitioned_count = serializers.IntegerField(help_text="Number of reports whose state was changed.")
    skipped_count = serializers.IntegerField(help_text="Number of reports whose transition was not allowed.")
    failed_count = serializers.IntegerField(help_text="Number of reports that failed on invalid request data.")
    not_found_count = serializers.IntegerField(help_text="Number of requested ids not visible to the caller.")


# The thumbs rating at the end of the report body carries an optional note, capped in the UI at the
# same length as the dismissal note.
SIGNAL_REPORT_FEEDBACK_NOTE_MAX_LENGTH = SIGNAL_REPORT_DISMISSAL_NOTE_MAX_LENGTH


class SignalReportFeedbackRequestSerializer(serializers.Serializer):
    sentiment = serializers.ChoiceField(
        choices=[("positive", "positive"), ("negative", "negative")],
        help_text="The rating left on the report: 'positive' (thumbs up) or 'negative' (thumbs down).",
    )
    note = serializers.CharField(
        max_length=SIGNAL_REPORT_FEEDBACK_NOTE_MAX_LENGTH,
        help_text=(
            "Free-form note explaining the rating. Capped at "
            f"{SIGNAL_REPORT_FEEDBACK_NOTE_MAX_LENGTH} characters. Only submitted alongside a note — a "
            "bare thumb carries none — and, for a report authored by a scout, forwarded to that scout "
            "as a steering note."
        ),
    )


class SignalReportFeedbackResponseSerializer(serializers.Serializer):
    forwarded = serializers.BooleanField(
        help_text=(
            "Whether the note was forwarded to the report's authoring scout as a steering note. False "
            "when the report has no resolvable authoring scout, or the caller lacks scout-steering access."
        ),
    )


# Whole PR-refund feature gate. Checked server-side (not just in the UI) and keyed on the
# organization, matching the refund window (the org's billing period).
SIGNALS_PR_REFUNDS_FEATURE_FLAG = "signals-pr-refunds"


# User-facing message for each shared `refund_ineligibility_reason` the refund action rejects on
# (`already_refunded` never reaches a 400 — it takes the idempotent 200 path instead).
_REFUND_INELIGIBLE_MESSAGES = {
    REFUND_INELIGIBLE_BILLING_EXEMPT: "This report is marked never-billable, so there is nothing to refund.",
    REFUND_INELIGIBLE_NO_BILLABLE_PR: "This report has no billable implementation PR to refund.",
    REFUND_INELIGIBLE_OUT_OF_PERIOD: "This PR was billed in a previous billing period and can no longer be refunded.",
}


class SignalReportRefundRequestSerializer(serializers.Serializer):
    reason = serializers.ChoiceField(
        choices=SignalReportRefund.Reason.choices,
        help_text=(
            "Why this PR is being refunded. One of: pr_incorrect (the PR doesn't address what the "
            "report promised), pr_not_useful (technically fine but not worth paying for), duplicate "
            "(covers work already charged elsewhere), other. Required — refund reviews key on it."
        ),
    )
    note = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=SIGNAL_REPORT_DISMISSAL_NOTE_MAX_LENGTH,
        help_text=(
            "Optional free-form context for the refund; stored on the refund and echoed in the "
            "report's dismissal artefact. Capped at 4000 characters."
        ),
    )


class SignalReportRefundResponseSerializer(SignalReportRefundSerializer):
    already_refunded = serializers.BooleanField(
        read_only=True,
        default=False,
        help_text=(
            "True when the report already had a refund and that existing refund is returned "
            "unchanged — refunds are one-per-report and repeat calls are idempotent."
        ),
    )

    class Meta(SignalReportRefundSerializer.Meta):
        fields = [*SignalReportRefundSerializer.Meta.fields, "already_refunded"]
        read_only_fields = fields


class SignalReportRefundSummaryResponseSerializer(serializers.Serializer):
    credited_refund_count = serializers.IntegerField(
        help_text=(
            "Number of credited-path refunds across the whole organization whose refunded PR run "
            "falls in the current billing period. Excluded-path refunds never reach billing usage, "
            "so they are deliberately absent."
        ),
    )
    credited_credits = serializers.IntegerField(
        help_text=(
            "Total signals credits those refunds returned (1 credit = $0.01). Divide by the flat "
            "per-PR charge to get the number of PRs to subtract from billing usage."
        ),
    )
    period_billable_credits = serializers.IntegerField(
        help_text=(
            "The organization's live billable signals credits for the current billing period, "
            "computed by the same rules as the nightly usage report — including PRs created today "
            "that billing hasn't recorded yet, and already excluding refund-excluded and "
            "billing-exempt reports. Take the max of this and billing's recorded usage for a live "
            "PR count that reacts to new PRs and same-day refunds immediately."
        ),
    )
    quota_limited = serializers.BooleanField(
        help_text=(
            "Whether autonomous PR generation is currently paused for this project because the "
            "organization is over its self-driving credits quota. Read from the quota limiter, so it "
            "reflects the same state the pipeline gates enforce."
        ),
    )


class SignalReportContentUpdateSerializer(serializers.Serializer):
    """Editable human-facing fields on a signal report (PATCH).

    Both fields are optional so a caller can change either independently, but at least one
    must be supplied. Every other report field — status, weights, judgments — is owned by the
    signals pipeline and is deliberately not writable here.
    """

    # min_length=1 (not just allow_blank=False) so the non-empty constraint surfaces as
    # `minLength: 1` in the generated OpenAPI/Zod schema — otherwise clients only learn an
    # empty string is invalid when the server rejects it.
    title = serializers.CharField(
        required=False,
        allow_blank=False,
        trim_whitespace=True,
        min_length=1,
        max_length=SIGNAL_REPORT_TITLE_MAX_LENGTH,
        help_text="New human-facing title for the report. Omit to leave the title unchanged.",
    )
    summary = serializers.CharField(
        required=False,
        allow_blank=False,
        trim_whitespace=True,
        min_length=1,
        max_length=SIGNAL_REPORT_SUMMARY_MAX_LENGTH,
        help_text=(
            "New summary (the report's description) explaining what the report is about. "
            "Omit to leave the summary unchanged."
        ),
    )

    def validate(self, attrs: dict) -> dict:
        if "title" not in attrs and "summary" not in attrs:
            raise serializers.ValidationError("Provide at least one of 'title' or 'summary' to update.")
        return attrs


@extend_schema_view(
    destroy=extend_schema(exclude=True),
)
class SignalReportViewSet(
    TeamAndOrgViewSetMixin,
    mixins.RetrieveModelMixin,
    mixins.ListModelMixin,
    mixins.DestroyModelMixin,
    viewsets.GenericViewSet,
):
    serializer_class = SignalReportSerializer
    authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication]
    permission_classes = [IsAuthenticated, APIScopePermission]
    scope_object = "task"
    queryset = SignalReport.objects.all()
    # Shared Q for "ready but not actionable" — used in status ranking and suggested-reviewer suppression.
    # Requires `latest_actionability_value` annotation to be applied first.
    _Q_READY_NOT_ACTIONABLE = Q(status=SignalReport.Status.READY) & Q(latest_actionability_value="not_actionable")
    _DEFAULT_SIGNAL_REPORT_ORDERING = "-is_suggested_reviewer,status,-updated_at"
    _SIGNAL_REPORT_ORDERING_FIELDS: dict[str, str] = {
        "status": "pipeline_status_rank",
        "is_suggested_reviewer": "is_suggested_reviewer",
        "signal_count": "signal_count",
        "total_weight": "total_weight",
        "priority": "priority_rank",
        "created_at": "created_at",
        "updated_at": "updated_at",
        "id": "id",
    }

    def safely_get_queryset(self, queryset):
        qs = queryset
        qs = self._scope_signal_report_queryset(qs)
        qs = self._exclude_deleted_signal_reports(qs)
        qs = self._apply_signal_report_status_filter(qs)
        qs = self._apply_signal_report_search_filter(qs)
        qs = self._apply_signal_report_source_product_filter(qs)
        qs = self._apply_signal_report_source_id_filter(qs)
        qs = self._apply_signal_report_scout_filter(qs)
        qs = self._apply_signal_report_scout_prefix_filter(qs)
        qs = self._apply_signal_report_implementation_pr_filter(qs)
        qs = self._apply_signal_report_suggested_reviewer_filter(qs)
        qs = self._apply_signal_report_task_filter(qs)
        qs = self._annotate_latest_actionability_value(qs)
        qs = self._apply_signal_report_actionability_filter(qs)
        qs = self._annotate_signal_report_status_rank(qs)
        qs = self._annotate_signal_report_priority(qs)
        qs = self._apply_signal_report_priority_filter(qs)
        qs = self._prefetch_signal_report_priority_artefacts(qs)
        qs = self._annotate_is_suggested_reviewer(qs)
        # Batched billable-moment lookup for the serializer's refund_ineligibility_reason field.
        qs = annotate_first_billable_pr_run_at(qs)
        if self.action != "list":
            qs = self._annotate_implementation_pr_url(qs)
        return qs

    def _scope_signal_report_queryset(self, queryset):
        # Count via a correlated subquery instead of `Count("artefacts")`,
        # so the main query doesn't LEFT JOIN + GROUP BY the full artefact table
        artefact_count_subquery = Subquery(
            SignalReportArtefact.objects.filter(report_id=OuterRef("id"))
            .values("report_id")
            .annotate(count=Count("*"))
            .values("count"),
            output_field=IntegerField(),
        )
        # select_related("refund"): the serializer renders the reverse OneToOne inline.
        return (
            queryset.filter(team=self.team)
            .select_related("refund")
            .annotate(
                artefact_count=Coalesce(artefact_count_subquery, Value(0), output_field=IntegerField()),
            )
        )

    def _exclude_deleted_signal_reports(self, queryset):
        # Deleted reports are terminal -- exclude from all endpoints (detail, list, actions)
        return queryset.exclude(status=SignalReport.Status.DELETED)

    # `deleted` is in the model but always stripped upstream by `_exclude_deleted_signal_reports`,
    # so it is never a valid filter target.
    _FILTERABLE_STATUSES = frozenset(SignalReport.Status.values) - {SignalReport.Status.DELETED}

    # Actions allowed to resolve a suppressed report by ID even without an explicit
    # `status` filter. These are the read/reopen paths the inbox's Dismissed tab needs:
    # `state` reopens a dismissed report, `retrieve` loads its detail, and `signals`
    # loads its evidence. `bulk_state` is included so a bulk restore (state='potential')
    # can reach suppressed reports too. `refund` is included so an already-archived but
    # billed report can still be refunded, and `feedback` because the detail view the
    # Dismissed tab renders ends in the thumbs rating, which must be able to forward its
    # note for the report it is displayed on. Mutating-by-ID actions (delete, reingest) are
    # deliberately NOT here, so a suppressed report stays unreachable for those and keeps
    # returning 404 — matching the existing contract.
    _SUPPRESSED_VISIBLE_ACTIONS = frozenset({"state", "bulk_state", "retrieve", "signals", "refund", "feedback"})

    # Human-readable explanation per bulk outcome, surfaced in each result's `detail` field
    # (transitioned needs none — its `status` already says where the report landed).
    _BULK_STATE_OUTCOME_DETAIL = {
        SignalReportBulkStateOutcome.SKIPPED: "This transition is not allowed from the report's current status.",
        SignalReportBulkStateOutcome.FAILED: "The request data was invalid for this report.",
        SignalReportBulkStateOutcome.NOT_FOUND: "No report with this id is visible to you.",
    }

    def _apply_signal_report_status_filter(self, queryset):
        status_filter = self.request.query_params.get("status")
        if status_filter:
            statuses = [s.strip() for s in status_filter.split(",") if s.strip()]
            invalid = [s for s in statuses if s not in self._FILTERABLE_STATUSES]
            if invalid:
                accepted = ", ".join(sorted(self._FILTERABLE_STATUSES))
                raise serializers.ValidationError(
                    {
                        "status": f"Invalid status value(s): {', '.join(sorted(set(invalid)))}. Accepted values: {accepted}."
                    }
                )
            return queryset.filter(status__in=statuses)
        # A few read/reopen actions must be able to reach a suppressed report by ID
        # (e.g. `state` reopens a dismissed report, `retrieve`/`signals` back the
        # inbox's Dismissed-tab detail view). Everywhere else — including the list and
        # mutating-by-ID actions like delete/reingest — suppressed reports stay hidden
        # unless an explicit `status` filter asks for them, or the caller opts out of
        # the default exclusions with `include_all_statuses=true` (agents deduping
        # against the full inbox state, human dismissals included, without enumerating
        # every status — and without breaking if the status set evolves).
        if self.action in self._SUPPRESSED_VISIBLE_ACTIONS:
            return queryset
        if self._include_all_statuses_requested():
            return queryset
        return queryset.exclude(status=SignalReport.Status.SUPPRESSED)

    def _include_all_statuses_requested(self) -> bool:
        # List-only: the flag widens the *list* for full-inbox-state scans (agent dedup). By-ID
        # actions ignore it entirely, so mutating actions like delete/reingest keep their existing
        # contract of 404ing on suppressed reports.
        if self.action != "list":
            return False
        raw = self.request.query_params.get("include_all_statuses")
        if raw is None or not raw.strip():
            return False
        value = raw.strip().lower()
        if value in ("1", "true", "yes"):
            return True
        if value in ("0", "false", "no"):
            return False
        raise serializers.ValidationError({"include_all_statuses": f"Invalid value: {raw!r}. Allowed: true, false."})

    def _apply_signal_report_search_filter(self, queryset):
        search = self.request.query_params.get("search")
        if not search:
            return queryset
        return queryset.filter(Q(title__icontains=search) | Q(summary__icontains=search))

    def _apply_signal_report_source_product_filter(self, queryset):
        source_product_filter = self.request.query_params.get("source_product")
        if not source_product_filter:
            return queryset
        # `source_id` already implies its product, so its narrower lookup subsumes this one. Skip
        # rather than run a second Datastore query for a strictly wider set.
        if self.request.query_params.get("source_id"):
            return queryset

        source_products = [s.strip() for s in source_product_filter.split(",") if s.strip()]
        if not source_products:
            return queryset

        report_ids_with_source = fetch_report_ids_for_source_products(self.team, source_products)
        return queryset.filter(id__in=report_ids_with_source)

    def _apply_signal_report_source_id_filter(self, queryset):
        """Reports a specific source record contributed to, e.g. one support ticket's reports.

        The owning product asks with the id it already has, instead of reaching into signals.

        Requires a single `source_product`, because a source id is only unique within one: emitters
        pass through the external system's own id, so GitHub issue 42 and Jira issue 42 both arrive as
        `"42"`. `SignalEmissionRecord` says the same thing with its `(team, source_product, source_type,
        source_id)` constraint. Without the product this would quietly mix products together.
        """
        source_id_filter = self.request.query_params.get("source_id")
        if not source_id_filter:
            return queryset

        source_ids = [s.strip() for s in source_id_filter.split(",") if s.strip()]
        if not source_ids:
            return queryset

        source_product = self.request.query_params.get("source_product")
        product = source_product.strip() if source_product else ""
        if not product or "," in product:
            raise exceptions.ValidationError(
                {
                    "source_id": "Pass exactly one source_product alongside source_id. A source id is only "
                    "unique within its product, so filtering without one would mix products together."
                }
            )
        by_source = fetch_live_report_ids_for_source_ids(self.team, source_ids, product)
        report_ids = {report_id for ids in by_source.values() for report_id in ids}
        return queryset.filter(id__in=report_ids)

    def _apply_signal_report_scout_filter(self, queryset):
        scout_filter = self.request.query_params.get("scout")
        if not scout_filter:
            return queryset

        scout_names = [s.strip() for s in scout_filter.split(",") if s.strip()]
        if not scout_names:
            return queryset

        report_ids_with_scout = fetch_report_ids_for_scout_names(self.team, scout_names)
        return queryset.filter(id__in=report_ids_with_scout)

    def _apply_signal_report_scout_prefix_filter(self, queryset):
        scout_prefix = (self.request.query_params.get("scout_prefix") or "").strip()
        if not scout_prefix:
            return queryset

        report_ids_with_prefix = fetch_report_ids_for_scout_prefix(self.team, scout_prefix)
        return queryset.filter(id__in=report_ids_with_prefix)

    def _latest_suggested_reviewers_qs(self):
        """`suggested_reviewers` rows that are the *current* (latest) version for the correlated
        outer report (`OuterRef("id")`).

        suggested_reviewers is append-only, so only the newest row is the live reviewer set —
        older versions remain as history and must not match. A row is current iff no newer row of
        the same type exists for its report.
        """
        has_newer = Exists(
            SignalReportArtefact.objects.filter(
                report_id=OuterRef("report_id"),
                type=SignalReportArtefact.ArtefactType.SUGGESTED_REVIEWERS,
                created_at__gt=OuterRef("created_at"),
            )
        )
        return SignalReportArtefact.objects.filter(
            report_id=OuterRef("id"),
            type=SignalReportArtefact.ArtefactType.SUGGESTED_REVIEWERS,
        ).filter(~has_newer)

    def _implementation_pr_report_filter(self):
        # Reports with a shipped implementation PR, as a `Q` on `SignalReport.id`. Decorrelated:
        # starts from the (small, index-backed) set of this team's tasks whose runs carry a non-empty
        # `pr_url` and maps them to reports via the indexed `task_id` columns — instead of a correlated
        # `Exists` over `tasks.TaskRun` evaluated once per candidate report (which made the inbox
        # PR-tab count scan the whole `ready` set per PR'd run).
        return SignalReport.reports_for_task_ids_filter(tasks_facade.task_ids_with_pr_url_subquery(self.team.id))

    def _apply_signal_report_implementation_pr_filter(self, queryset):
        # `has_implementation_pr=true|false` filters reports by whether a shipped
        # implementation PR exists. Lets the inbox count PR reports (the "Pull
        # requests" tab) with a cheap count query instead of paging the whole list
        # and filtering client-side. Absent or empty param leaves the list
        # unchanged; an unrecognized value is a 400.
        raw = self.request.query_params.get("has_implementation_pr")
        if raw is None or not raw.strip():
            return queryset
        value = raw.strip().lower()
        if value in ("1", "true", "yes"):
            wants_pr = True
        elif value in ("0", "false", "no"):
            wants_pr = False
        else:
            raise serializers.ValidationError(
                {"has_implementation_pr": f"Invalid value: {raw!r}. Allowed: true, false."}
            )
        pr_filter = self._implementation_pr_report_filter()
        return queryset.filter(pr_filter) if wants_pr else queryset.exclude(pr_filter)

    def _apply_signal_report_suggested_reviewer_filter(self, queryset):
        suggested_reviewer_filter = self.request.query_params.get("suggested_reviewers")
        if not suggested_reviewer_filter:
            return queryset

        reviewer_user_uuids = [s.strip() for s in suggested_reviewer_filter.split(",") if s.strip()]
        try:
            reviewer_user_uuids = [str(uuid.UUID(user_uuid)) for user_uuid in reviewer_user_uuids]
        except (ValueError, AttributeError) as e:
            raise serializers.ValidationError({"suggested_reviewers": f"Invalid user UUID: {e}"})

        reviewer_github_logins = list(
            get_org_member_github_logins_by_user_uuid(self.team.id, reviewer_user_uuids).values()
        )
        if not reviewer_github_logins:
            return queryset.none()

        reviewer_json_filters = [
            json.dumps([{"github_login": github_login}]) for github_login in reviewer_github_logins
        ]
        reviewer_where = " OR ".join(["content::jsonb @> %s::jsonb"] * len(reviewer_json_filters))
        return queryset.filter(
            Exists(
                # nosemgrep: python.django.security.audit.query-set-extra.avoid-query-set-extra (parameterized via params)
                self._latest_suggested_reviewers_qs().extra(
                    where=[reviewer_where],
                    params=reviewer_json_filters,
                )
            )
        )

    def _apply_signal_report_task_filter(self, queryset):
        # Reports a given task is associated with — used by running agents ("which reports am I
        # working against?") and by the agent harness to fan commit artefacts out to them. Uses the
        # unified association (task_run artefacts + legacy SignalReportTask); the team-scoped outer
        # queryset keeps the result within the project.
        task_filter = self.request.query_params.get("task_id")
        if not task_filter:
            return queryset
        try:
            task_uuid = uuid.UUID(task_filter.strip())
        except (ValueError, AttributeError) as e:
            raise serializers.ValidationError({"task_id": f"Invalid task UUID: {e}"})
        return queryset.filter(SignalReport.reports_for_task_filter(task_uuid))

    def _apply_signal_report_priority_filter(self, queryset):
        # Filters on the `priority_rank` annotation, which must be applied first.
        # Reports without a priority artefact (coalesced to "~") are excluded when this filter is set.
        priority_filter = self.request.query_params.get("priority")
        if not priority_filter:
            return queryset

        values = [p.strip().upper() for p in priority_filter.split(",") if p.strip()]
        if not values:
            return queryset

        allowed = set(AutonomyPriority.values)
        invalid = [v for v in values if v not in allowed]
        if invalid:
            raise serializers.ValidationError(
                {
                    "priority": f"Invalid priority value(s): {', '.join(sorted(set(invalid)))}. Allowed: {', '.join(sorted(allowed))}."
                }
            )

        return queryset.filter(priority_rank__in=values)

    def _apply_signal_report_actionability_filter(self, queryset):
        # Filters on the `latest_actionability_value` annotation (the actionability
        # choice from the latest actionability_judgment artefact), which must be
        # annotated first. Powers the inbox's actionability-keyed tabs: the Reports
        # tab passes the two actionable values, the staff-only Not-actionable tab
        # passes `not_actionable`. Reports without an actionability judgment
        # (annotation is NULL) are excluded when this filter is set. Absent or empty
        # param leaves the list unchanged; an unrecognized value is a 400.
        actionability_filter = self.request.query_params.get("actionability")
        if not actionability_filter:
            return queryset

        values = [a.strip() for a in actionability_filter.split(",") if a.strip()]
        if not values:
            return queryset

        allowed = {choice.value for choice in ActionabilityChoice}
        invalid = [v for v in values if v not in allowed]
        if invalid:
            raise serializers.ValidationError(
                {
                    "actionability": f"Invalid actionability value(s): {', '.join(sorted(set(invalid)))}. "
                    f"Allowed: {', '.join(sorted(allowed))}."
                }
            )

        return queryset.filter(latest_actionability_value__in=values)

    def _annotate_signal_report_status_rank(self, queryset):
        # `ordering=status` uses semantic stage rank (annotation), not lexicographic `status` column order.
        # `status=ready` splits into two virtual stages (requires `latest_actionability_value`):
        # 0 = ready + actionable (or no judgment yet), 1 = ready + not_actionable; then other stages.
        return queryset.annotate(
            pipeline_status_rank=Case(
                When(self._Q_READY_NOT_ACTIONABLE, then=Value(1)),
                When(status=SignalReport.Status.READY, then=Value(0)),
                When(status=SignalReport.Status.PENDING_INPUT, then=Value(2)),
                When(status=SignalReport.Status.IN_PROGRESS, then=Value(3)),
                When(status=SignalReport.Status.CANDIDATE, then=Value(4)),
                When(status=SignalReport.Status.POTENTIAL, then=Value(5)),
                When(status=SignalReport.Status.FAILED, then=Value(6)),
                When(status=SignalReport.Status.RESOLVED, then=Value(7)),
                When(status=SignalReport.Status.SUPPRESSED, then=Value(8)),
                When(status=SignalReport.Status.DELETED, then=Value(9)),
                default=Value(50),
                output_field=IntegerField(),
            )
        )

    def _annotate_signal_report_priority(self, queryset):
        # `ordering=priority` sorts by the priority value ("P0"–"P4") from the latest priority_judgment
        # artefact. These sort lexicographically, so we extract via jsonb and coalesce NULL to "~"
        # (sorts after "P4") for reports without a priority. The startswith guard skips non-object content.
        latest_priority = Subquery(
            SignalReportArtefact.objects.filter(
                report_id=OuterRef("id"),
                type=SignalReportArtefact.ArtefactType.PRIORITY_JUDGMENT,
                content__startswith="{",
            )
            .order_by("-created_at")
            .annotate(
                _priority_val=Func(
                    Cast(F("content"), output_field=JSONField()),
                    Value("priority"),
                    function="jsonb_extract_path_text",
                    output_field=CharField(),
                ),
            )
            .values("_priority_val")[:1],
            output_field=CharField(),
        )
        return queryset.annotate(
            priority_rank=Coalesce(latest_priority, Value("~"), output_field=CharField()),
        )

    def _annotate_latest_actionability_value(self, queryset):
        # Extract the "actionability" value from the latest actionability_judgment artefact.
        latest_actionability = Subquery(
            SignalReportArtefact.objects.filter(
                report_id=OuterRef("id"),
                type=SignalReportArtefact.ArtefactType.ACTIONABILITY_JUDGMENT,
                content__startswith="{",
            )
            .order_by("-created_at")
            .annotate(
                _actionability_val=Func(
                    Cast(F("content"), output_field=JSONField()),
                    Value("actionability"),
                    function="jsonb_extract_path_text",
                    output_field=CharField(),
                ),
            )
            .values("_actionability_val")[:1],
            output_field=CharField(),
        )
        return queryset.annotate(latest_actionability_value=latest_actionability)

    def _prefetch_signal_report_priority_artefacts(self, queryset):
        return queryset.prefetch_related(
            Prefetch(
                "artefacts",
                queryset=SignalReportArtefact.objects.filter(
                    type=SignalReportArtefact.ArtefactType.PRIORITY_JUDGMENT
                ).order_by("-created_at"),
                to_attr="prefetched_priority_artefacts",
            ),
            Prefetch(
                "artefacts",
                queryset=SignalReportArtefact.objects.filter(
                    type=SignalReportArtefact.ArtefactType.ACTIONABILITY_JUDGMENT
                ).order_by("-created_at"),
                to_attr="prefetched_actionability_artefacts",
            ),
            Prefetch(
                "artefacts",
                queryset=SignalReportArtefact.objects.filter(type=SignalReportArtefact.ArtefactType.DISMISSAL).order_by(
                    "-created_at"
                ),
                to_attr="prefetched_dismissal_artefacts",
            ),
        )

    def _annotate_is_suggested_reviewer(self, queryset):
        # Annotate is_suggested_reviewer by resolving the current user's GitHub login
        # and checking jsonb containment on the artefact content list. This stays fresh
        # even when a user connects their GitHub account after the report was generated.
        # Never true for ready + not_actionable — there is nothing actionable to review.
        # Failed reports are excluded too — pipelines that errored should not bubble as "needs your review".
        github_login = self._get_github_login(self.request.user)
        if not github_login:
            return queryset.annotate(is_suggested_reviewer=Value(False))

        # github_login comes from our own UserSocialAuth DB, not user input.
        suggested_exists = Exists(
            # nosemgrep: python.django.security.audit.query-set-extra.avoid-query-set-extra (parameterized via params)
            self._latest_suggested_reviewers_qs().extra(
                where=["content::jsonb @> %s::jsonb"],
                params=[json.dumps([{"github_login": github_login}])],
            )
        )
        return queryset.annotate(
            is_suggested_reviewer=Case(
                When(self._Q_READY_NOT_ACTIONABLE, then=Value(False)),
                When(status=SignalReport.Status.FAILED, then=Value(False)),
                default=suggested_exists,
                output_field=BooleanField(),
            ),
        )

    def _annotate_implementation_pr_url(self, queryset):
        # Latest TaskRun output->pr_url across the tasks associated with each report, unified over
        # the task_run artefact log + legacy SignalReportTask rows (see associated_task_runs_filter).
        # Only implementation runs carry a pr_url, so the non-empty-pr_url filter inside the facade
        # subquery makes "any associated task" resolve to the implementation PR.
        latest_impl_pr_url = tasks_facade.latest_task_run_pr_url_subquery(
            SignalReport.associated_task_runs_filter(OuterRef(OuterRef("id"))),
        )
        # Resolved over the same run, so the merge flag always describes the PR URL alongside it.
        latest_impl_pr_merged = tasks_facade.latest_task_run_pr_merged_subquery(
            SignalReport.associated_task_runs_filter(OuterRef(OuterRef("id"))),
        )
        return queryset.annotate(
            implementation_pr_url=latest_impl_pr_url,
            implementation_pr_merged=latest_impl_pr_merged,
        )

    def filter_queryset(self, queryset):
        queryset = super().filter_queryset(queryset)
        return self._apply_signal_report_ordering(queryset)

    def _parse_ordering_string(self, raw: str) -> list[str]:
        parts = [p.strip() for p in raw.split(",") if p.strip()]
        clauses: list[str] = []
        for part in parts:
            descending = part.startswith("-")
            name = part[1:] if descending else part
            db_field = self._SIGNAL_REPORT_ORDERING_FIELDS.get(name)
            if db_field is None:
                continue
            clause = f"-{db_field}" if descending else db_field
            clauses.append(clause)
        return clauses

    @property
    def _default_signal_report_ordering_clauses(self) -> list[str]:
        return self._parse_ordering_string(self._DEFAULT_SIGNAL_REPORT_ORDERING)

    def _parse_signal_report_ordering(self) -> list[str]:
        raw = self.request.query_params.get("ordering", self._DEFAULT_SIGNAL_REPORT_ORDERING)
        if not raw or not str(raw).strip():
            return self._default_signal_report_ordering_clauses
        clauses = self._parse_ordering_string(str(raw).strip())
        return clauses if clauses else self._default_signal_report_ordering_clauses

    def _apply_signal_report_ordering(self, queryset):
        clauses = self._parse_signal_report_ordering()
        has_id = any((c[1:] if c.startswith("-") else c) == "id" for c in clauses)
        if not has_id:
            clauses = [*clauses, "id"]
        return queryset.order_by(*clauses)

    @staticmethod
    def _get_github_login(user) -> str | None:
        login = user.get_github_login()
        return login.lower() if login else None

    def get_serializer_context(self):
        return {
            **super().get_serializer_context(),
            "team": self.team,
            # For the serializer's refund_ineligibility_reason field — computed once per request,
            # not per report (the bounds are org-level).
            "billing_period_bounds": current_billing_period_bounds(self.organization),
        }

    def _enriched_report_context(self, report: SignalReport) -> dict:
        # Detail-view parity with list(): inject the source-product and PR-url maps the
        # SignalReportSerializer reads, so single-report responses aren't silently degraded.
        # Both lookups are best-effort: the serializer degrades to empty values when a map
        # is missing, so a Datastore/Postgres hiccup must not turn an otherwise-available
        # report (or an already-committed state change) into a 500.
        report_ids = [str(report.id)]
        try:
            signal_meta_map = fetch_source_products_for_reports(self.team, report_ids)
        except Exception:
            logger.exception("signals.enriched_context.source_products_failed", report_id=str(report.id))
            signal_meta_map = {}
        try:
            implementation_pr_by_report = fetch_implementation_pr_state_for_reports(report_ids)
        except Exception:
            logger.exception("signals.enriched_context.implementation_pr_url_failed", report_id=str(report.id))
            implementation_pr_by_report = {}
        return {
            **self.get_serializer_context(),
            "source_products_map": {rid: meta.source_products for rid, meta in signal_meta_map.items()},
            "scout_names_map": {rid: meta.scout_name for rid, meta in signal_meta_map.items() if meta.scout_name},
            "implementation_pr_url_map": {rid: pr.url for rid, pr in implementation_pr_by_report.items()},
            "implementation_pr_merged_ids": {rid for rid, pr in implementation_pr_by_report.items() if pr.merged},
        }

    def retrieve(self, request, *args, **kwargs):
        report = self.get_object()
        serializer = self.get_serializer(report, context=self._enriched_report_context(report))
        return Response(serializer.data)

    @validated_request(
        request_serializer=SignalReportContentUpdateSerializer,
        responses={
            200: OpenApiResponse(response=SignalReportSerializer, description="Report updated."),
            400: OpenApiResponse(description="Neither title nor summary supplied, or a value failed validation."),
            404: OpenApiResponse(description="Report not found for this project."),
        },
        summary="Edit a report's title or summary",
        description=(
            "Edit the human-facing title and/or summary (description) of a signal report, addressed "
            "by id. Both fields are optional — supply only the ones you want to change; at least one "
            "is required. Every other report field (status, weights, judgments) is managed by the "
            "signals pipeline and cannot be set here. Returns the full updated report."
        ),
        operation_id="signals_reports_partial_update",
    )
    def partial_update(self, request: ValidatedRequest, *args, **kwargs) -> Response:
        report = cast(SignalReport, self.get_object())
        data = request.validated_data
        # Attribution mirrors the other artefact-writing paths (suggested reviewers, commit/task_run):
        # the edit is the agent's task when an `X-Insights-Task-Id` header is present, otherwise the
        # requesting user. Resolved up front so a bad task header 400s before we mutate anything.
        attribution = resolve_request_attribution(request, self.team.id)
        update_fields: list[str] = []
        # Each real change is logged as its own append-only edit-history artefact capturing the
        # before/after, so the report carries an audit trail of human/agent title & summary edits.
        edit_artefacts: list[TitleChange | SummaryChange] = []
        if "title" in data and data["title"] != report.title:
            edit_artefacts.append(TitleChange(old_title=report.title, new_title=data["title"]))
            report.title = data["title"]
            update_fields.append("title")
        if "summary" in data and data["summary"] != report.summary:
            edit_artefacts.append(SummaryChange(old_summary=report.summary, new_summary=data["summary"]))
            report.summary = data["summary"]
            update_fields.append("summary")

        if update_fields:
            # `updated_at` is auto_now, but `update_fields` saves only the listed columns, so add it
            # explicitly to keep the edit timestamped.
            update_fields.append("updated_at")
            # This text has not been through the safety judge, and the report's existing verdict was
            # reached on the text this edit replaces. Marking the save retracts the report's embedding
            # rather than indexing unreviewed content under a stale approval (see receivers.py).
            report._unreviewed_edit = True  # type: ignore[attr-defined]
            with transaction.atomic():
                report.save(update_fields=update_fields)
                for content in edit_artefacts:
                    SignalReportArtefact.add_log(
                        team_id=self.team.id,
                        report_id=str(report.id),
                        content=content,
                        attribution=attribution,
                    )
        return Response(SignalReportSerializer(report, context=self._enriched_report_context(report)).data)

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="status",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of statuses to include. "
                    "Valid values: potential, candidate, in_progress, pending_input, ready, resolved, failed, suppressed. "
                    "Defaults to all statuses except suppressed."
                ),
            ),
            OpenApiParameter(
                name="include_all_statuses",
                type=OpenApiTypes.BOOL,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "When true, the list includes reports in every status with no default exclusions applied — "
                    "currently that adds suppressed (dismissed) reports, which are otherwise hidden. Use it to "
                    "see the full inbox state (e.g. deduplicating before creating a report) and read each row's "
                    "status (plus dismissal_reason/dismissal_note on dismissed rows) before acting. Deleted "
                    "reports are terminal and never returned. Defaults to false, which keeps the existing "
                    "default exclusions. Ignored when an explicit 'status' filter is set — that filter alone "
                    "decides which statuses are returned."
                ),
            ),
            OpenApiParameter(
                name="search",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Case-insensitive substring match against report title and summary.",
            ),
            OpenApiParameter(
                name="source_product",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of source products to include. Reports are kept if at least one of "
                    "their contributing signals comes from one of these products (e.g. error_tracking, session_replay)."
                ),
            ),
            OpenApiParameter(
                name="source_id",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of source record ids. Reports are kept if at least one of their "
                    "contributing signals came from one of these records — e.g. pass a support ticket's UUID to "
                    "see what the inbox already found for that ticket. Requires exactly one source_product, "
                    "since a source id is only unique within its product."
                ),
            ),
            OpenApiParameter(
                name="scout",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of scout skill_name slugs (e.g. signals-scout-error-tracking). "
                    "Reports are kept if at least one of their contributing signals was authored by one of "
                    "these scouts. Combines with source_product as an AND."
                ),
            ),
            OpenApiParameter(
                name="scout_prefix",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Scout skill_name prefix (e.g. signals-scout-customer-analytics). Reports are kept if at "
                    "least one of their contributing signals was authored by a scout whose skill_name starts "
                    "with this prefix — new scouts in the family match without callers listing every name. "
                    "Combines with the other filters as an AND."
                ),
            ),
            OpenApiParameter(
                name="suggested_reviewers",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of Insights user UUIDs. Reports are kept if their suggested reviewers "
                    "include any of the given users."
                ),
            ),
            OpenApiParameter(
                name="priority",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated list of priorities to include. Valid values: P0, P1, P2, P3, P4. "
                    "Reports without a priority assignment are excluded when this filter is set."
                ),
            ),
            OpenApiParameter(
                name="ordering",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Comma-separated ordering clauses. Each clause is a field name optionally prefixed with '-' "
                    "for descending. Allowed fields: status, is_suggested_reviewer, signal_count, total_weight, "
                    "priority, created_at, updated_at, id. Defaults to '-is_suggested_reviewer,status,-updated_at'."
                ),
            ),
            OpenApiParameter(
                name="task_id",
                type=OpenApiTypes.UUID,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Only reports associated with this task (via the report's task associations).",
            ),
            OpenApiParameter(
                name="has_implementation_pr",
                type=OpenApiTypes.BOOL,
                location=OpenApiParameter.QUERY,
                required=False,
                description=(
                    "Filter reports by whether a shipped implementation pull request exists. "
                    "'true' keeps only reports with a PR; 'false' keeps only those without. "
                    "Pair with limit=1 to count PR reports cheaply."
                ),
            ),
        ],
    )
    @tracer.start_as_current_span("signals.reports.list")
    def list(self, request, *args, **kwargs):
        # The reports list is the primary inbox-load endpoint. Each phase gets its own child span
        # so a slow load can be attributed to Postgres (queryset annotations), Datastore (source
        # products), the task facade (PR urls), or serialization, rather than one opaque request.
        with tracer.start_as_current_span("signals.reports.list.queryset"):
            queryset = self.filter_queryset(self.get_queryset())
            page = self.paginate_queryset(queryset)
            reports = list(page if page is not None else queryset)

        report_ids = [str(r.id) for r in reports]
        trace.get_current_span().set_attribute("signals.reports.list.count", len(report_ids))

        # Both lookups are best-effort decorative metadata (source-product badges, scout names, PR
        # urls). The serializer degrades to empty values when a map is missing, so a Datastore or
        # backend hiccup in either must not 500 the whole inbox load — fall back to empty and log.
        with tracer.start_as_current_span("signals.reports.list.fetch_source_products"):
            try:
                signal_meta_map = fetch_source_products_for_reports(self.team, report_ids) if report_ids else {}
            except Exception:
                logger.exception("signals.reports.list.source_products_failed", report_count=len(report_ids))
                signal_meta_map = {}

        with tracer.start_as_current_span("signals.reports.list.fetch_implementation_pr_urls"):
            try:
                implementation_pr_by_report = fetch_implementation_pr_state_for_reports(report_ids)
            except Exception:
                logger.exception("signals.reports.list.implementation_pr_url_failed", report_count=len(report_ids))
                implementation_pr_by_report = {}

        context = {
            **self.get_serializer_context(),
            "source_products_map": {rid: meta.source_products for rid, meta in signal_meta_map.items()},
            "scout_names_map": {rid: meta.scout_name for rid, meta in signal_meta_map.items() if meta.scout_name},
            "implementation_pr_url_map": {rid: pr.url for rid, pr in implementation_pr_by_report.items()},
            "implementation_pr_merged_ids": {rid for rid, pr in implementation_pr_by_report.items() if pr.merged},
        }
        serializer = self.get_serializer(reports, many=True, context=context)

        with tracer.start_as_current_span("signals.reports.list.serialize"):
            data = serializer.data

        if page is not None:
            return self.get_paginated_response(data)
        return Response(data)

    @extend_schema(exclude=True)
    @action(detail=False, methods=["get"], url_path="available_reviewers", required_scopes=["task:read"])
    def available_reviewers(self, request, **kwargs):
        with tracer.start_as_current_span("signals.available_reviewers") as span:
            login_to_user = get_org_member_github_login_to_user_map(self.team.id) or {}
            query = (request.query_params.get("query") or "").strip().lower()

            users_by_uuid = {str(user.uuid): user for user in login_to_user.values()}

            candidate_count = len(users_by_uuid)
            span.set_attribute("signals.available_reviewers.candidate_count", candidate_count)

            # The full candidate list is returned unpaginated. If an org grows past the threshold,
            # report it (non-blocking) so we know to add pagination before the payload gets large.
            # `capture_exception` logs an exception and is not deduplicated, so we throttle to at
            # most one report per org per day via the cache — otherwise a >threshold org would log
            # on every popover open. (The span's candidate_count attribute is recorded every request
            # regardless, if a metric-based alert is preferred later.)
            if (
                not query
                and candidate_count > REVIEWER_PAGINATION_THRESHOLD
                and cache.add(f"signals:available_reviewers_over_threshold:{self.team.id}", True, 60 * 60 * 24)
            ):
                capture_exception(
                    Exception(
                        f"available_reviewers exceeded pagination threshold: {candidate_count} "
                        f"candidates > {REVIEWER_PAGINATION_THRESHOLD}; this endpoint should be paginated."
                    ),
                    additional_properties={
                        "team_id": self.team.id,
                        "candidate_count": candidate_count,
                        "threshold": REVIEWER_PAGINATION_THRESHOLD,
                    },
                )

            filtered_users = [
                (user_uuid, user)
                for user_uuid, user in users_by_uuid.items()
                if not query
                or query in f"{user.first_name} {user.last_name}".strip().lower()
                or query in (user.email or "").lower()
            ]

            reviewers = {
                user_uuid: {
                    "name": f"{user.first_name} {user.last_name}".strip(),
                    "email": user.email or "",
                }
                for user_uuid, user in sorted(
                    filtered_users,
                    key=lambda item: (
                        (item[1].first_name or "").lower(),
                        (item[1].last_name or "").lower(),
                        (item[1].email or "").lower(),
                        item[0],
                    ),
                )
            }

            span.set_attribute("signals.available_reviewers.result_count", len(reviewers))

            return Response(reviewers)

    @extend_schema(exclude=True)
    @action(detail=True, methods=["put"], url_path="reviewers", required_scopes=["task:write"])
    def reviewers(self, request, **kwargs):
        """Set a report's suggested reviewers (full-replacement PUT), whether or not the report already
        has any. Appends a new latest-wins `suggested_reviewers` status row — the same write the artefact
        PUT performs, but addressed by report so a report with zero reviewers (and thus no artefact yet)
        can still be assigned one. App-only: agents append reviewers via the artefacts POST instead."""
        report = cast(SignalReport, self.get_object())

        write_serializer = SignalReportArtefactWriteSerializer(data=request.data)
        write_serializer.is_valid(raise_exception=True)
        entries = write_serializer.validated_data["content"]

        try:
            new_artefact, seen = append_suggested_reviewers(
                team=self.team,
                report_id=str(report.id),
                entries=entries,
                request=request,
            )
        except ReviewerWriteError as e:
            return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)

        # Return the read-shape (enriched) so the client sees the canonical result, matching the artefact PUT.
        login_map = resolve_org_github_login_to_users(self.team.id, list(seen)) if seen else {}
        read_serializer = SignalReportArtefactSerializer(
            new_artefact,
            context={
                **self.get_serializer_context(),
                "signals_github_login_to_user_map": login_map,
            },
        )
        return Response(read_serializer.data)

    def destroy(self, request, *args, **kwargs):
        """Soft-delete a report and its signals via the deletion workflow."""
        report = cast(SignalReport, self.get_object())
        report_id = str(report.id)
        team_id = self.team.id

        try:
            client = sync_connect()
            async_to_sync(client.start_workflow)(  # type: ignore
                "signal-report-deletion",  # type: ignore
                SignalReportDeletionWorkflowInputs(team_id=team_id, report_id=report_id),  # type: ignore
                id=SignalReportDeletionWorkflow.workflow_id_for(team_id, report_id),
                task_queue=settings.VIDEO_EXPORT_TASK_QUEUE,
                execution_timeout=timedelta(minutes=30),
                retry_policy=RetryPolicy(maximum_attempts=1),
            )
        except WorkflowAlreadyStartedError:
            return Response({"status": "already_running", "report_id": report_id}, status=status.HTTP_200_OK)
        except Exception:
            logger.exception("Failed to start deletion workflow for report %s", report_id)
            return Response(
                {"error": "Failed to start deletion workflow."},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )

        # Hide the report from the list immediately while signal deletion continues asynchronously.
        updated_fields = report.transition_to(SignalReport.Status.DELETED)
        report.save(update_fields=updated_fields)

        return Response({"status": "deletion_started", "report_id": report_id}, status=status.HTTP_202_ACCEPTED)

    @extend_schema(
        summary="List a report's signals",
        description="Fetch all signals for a report from Datastore, including full metadata.",
        responses={200: ReportSignalsResponseSerializer},
    )
    @action(detail=True, methods=["get"], url_path="signals", required_scopes=["task:read"])
    def signals(self, request, pk=None, **kwargs):
        """Fetch all signals for a report from Datastore, including full metadata."""
        report = self.get_object()
        report_data = SignalReportSerializer(report, context=self._enriched_report_context(report)).data
        signals_list = fetch_signals_for_report_sync(self.team, str(report.id))
        return Response({"report": report_data, "signals": signals_list})

    @extend_schema(
        request=SignalReportStateRequestSerializer,
        responses={200: SignalReportSerializer},
    )
    @action(detail=True, methods=["post"], url_path="state", required_scopes=["task:write"])
    def state(self, request, pk=None, **kwargs):
        """
        Transition a report to a new state. The model validates allowed transitions.

        The request body is validated by SignalReportStateRequestSerializer — only the
        fields it declares (state, dismissal_reason, dismissal_note, snooze_for) are read,
        and only snooze_for is ever forwarded to transition_to. Any other key is ignored,
        so internal transition_to kwargs (reset_weight, error, ...) can't be injected.

        Body: {
            "state": "suppressed" | "potential" | "resolved",
            # Optional dismissal feedback (honored when state == "suppressed", "potential", or "resolved"):
            "dismissal_reason": "<canonical reason code, see SIGNAL_REPORT_DISMISSAL_REASON_CHOICES>",
            "dismissal_note": "free-form text",
            # Optional, only honored for state == "potential":
            "snooze_for": <number of additional signals before re-promotion>,
        }
        """
        report = cast(SignalReport, self.get_object())

        serializer = SignalReportStateRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        outcome, detail = self._transition_report_state(
            report,
            target=data["state"],
            dismissal_reason=data.get("dismissal_reason"),
            dismissal_note=data.get("dismissal_note"),
            snooze_for=data.get("snooze_for"),
        )

        if outcome == SignalReportBulkStateOutcome.SKIPPED:
            return Response(
                {"error": detail or "Invalid state transition for this report."},
                status=status.HTTP_409_CONFLICT,
            )
        if outcome == SignalReportBulkStateOutcome.FAILED:
            return Response(
                {"error": detail or "Invalid data for state transition."},
                status=status.HTTP_400_BAD_REQUEST,
            )

        self._forward_dismissal_note(
            reports=[report],
            dismissal_reason=data.get("dismissal_reason"),
            dismissal_note=data.get("dismissal_note"),
        )

        return Response(SignalReportSerializer(report, context=self._enriched_report_context(report)).data)

    @extend_schema(
        summary="Leave feedback on a report",
        description=(
            "Record a note left with the thumbs rating at the end of a report. The rating itself is a "
            "product-analytics event; this endpoint exists to carry the note into the scout steering "
            "channel. For a report authored by a scout, the note is forwarded to that scout as a "
            "steering note it reads on its next run; for any other report there is nothing to steer and "
            "the call is a no-op success. The report's state is never changed."
        ),
        request=SignalReportFeedbackRequestSerializer,
        responses={200: SignalReportFeedbackResponseSerializer},
    )
    @action(detail=True, methods=["post"], url_path="feedback", required_scopes=["task:write"])
    def feedback(self, request, pk=None, **kwargs):
        """Forward a report's feedback note to the scout that authored it.

        Feedback-only: unlike `state`, this never transitions the report. The note is a derived
        convenience — the durable record of the rating is the analytics event the client fires —
        so forwarding is best-effort and never fails the request.
        """
        report = cast(SignalReport, self.get_object())

        serializer = SignalReportFeedbackRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        note_id = forward_feedback_note(
            team=self.team,
            report_id=str(report.id),
            sentiment=data["sentiment"],
            note=data["note"],
            user_id=request.user.id if isinstance(request.user, User) else None,
            scoped_team_ids=get_authenticator_scoped_team_ids(request.successful_authenticator),
            api_scopes=get_authenticator_scopes(request.successful_authenticator),
        )
        return Response(SignalReportFeedbackResponseSerializer({"forwarded": note_id is not None}).data)

    def _forward_dismissal_note(
        self,
        *,
        # `Sequence`, not `list`: this class defines a `list` action, which shadows the builtin for
        # annotations evaluated in the class body at import time.
        reports: Sequence[SignalReport],
        dismissal_reason: str | None,
        dismissal_note: str | None,
    ) -> None:
        """Leave the caller's dismissal note where scout runs actually read it.

        Called once per request with every report that transitioned, so one note applied to a bulk
        dismissal reaches each affected scout once instead of once per report. Runs after the
        transitions have committed, because the note is derived context and the `dismissal` artefact
        written alongside each transition remains the record of the feedback. Forwarding resolves
        the reports' resulting status itself, drops the transitions this channel has nothing to say
        about, and authorizes the caller against the scout-note write gates, so this call site only
        hands it the request's principal.
        """
        forward_dismissal_note(
            team=self.team,
            reports=reports,
            reason=dismissal_reason,
            note=dismissal_note,
            request=self.request,
        )

    def _request_attribution(self) -> ArtefactAttribution:
        """Attribution for this request, resolved once and reused.

        `resolve_request_attribution` validates the `X-Insights-Task-Id` header with a DB lookup, and
        both the request and the team are invariant across a call — so bulk_state, which transitions
        up to the per-call id cap in a loop, would otherwise repeat the identical query per report.
        """
        if not hasattr(self, "_cached_request_attribution"):
            self._cached_request_attribution = resolve_request_attribution(self.request, self.team.id)
        return self._cached_request_attribution

    def _transition_report_state(
        self,
        report: SignalReport,
        *,
        target: str,
        dismissal_reason: str | None,
        dismissal_note: str | None,
        snooze_for: int | None,
    ) -> "tuple[SignalReportBulkStateOutcome, str | None]":
        """
        Apply one report state transition (plus optional dismissal artefact) and return a
        compact `(outcome, detail)` pair — `detail` is a caller-facing explanation for
        non-transitioned outcomes when the generic per-outcome message isn't enough. Shared by
        the single `state` action and the bulk `bulk_state` action so both honour the same
        restore/snooze semantics and transition guards. Expected invalid transitions are
        returned as outcomes (never raised) so a bulk run can record a per-id result and keep
        going.

        Only `snooze_for` (on a snooze back to "potential") is caller-controllable. Every other
        `transition_to` kwarg (signals_at_run_increment, reset_weight, title, summary, error) is an
        internal pipeline concern and must never be reachable from this public API surface, so it is
        passed explicitly rather than splatting caller-supplied kwargs.
        """
        target_status = SignalReport.Status(target)

        with transaction.atomic():
            # Lock the report row for the whole guard + transition so it serializes with the refund
            # path (which locks the same row before suppressing). Re-read committed state under the
            # lock: a resolve/restore that raced a refund must observe the refund's SUPPRESSED status
            # and refund row and refuse, instead of overwriting the suppression and re-opening
            # already-refunded work.
            if SignalReport.objects.select_for_update().filter(id=report.id, team_id=self.team.id).first() is not None:
                report.refresh_from_db()
            is_refunded = SignalReportRefund.objects.filter(report_id=report.id).exists()

            # Refund guard: a refunded report can never be billed again (see billing.py), so putting
            # it back where the pipeline can pick it up again is repeatable free work. POTENTIAL is
            # the live status — a potential report re-promotes to candidate on new signals and can
            # spawn a fresh research run (grouping.py). The refund is final, so any transition of a
            # refunded report back to POTENTIAL is refused, whether it currently sits in SUPPRESSED
            # (refund's own archive step) or in RESOLVED (a refunded merged-PR report stays resolved,
            # so the guard can't key on SUPPRESSED alone). RESOLVED is terminal and never re-promotes,
            # so resolving a refunded report is refused only out of the archive, to keep the refund's
            # suppression — and the PR close it triggers — from being undone.
            if is_refunded and (
                target_status == SignalReport.Status.POTENTIAL
                or (report.status == SignalReport.Status.SUPPRESSED and target_status == SignalReport.Status.RESOLVED)
            ):
                return SignalReportBulkStateOutcome.SKIPPED, "Refunded reports can't be restored."

            # Archiving must not grant a transition the report couldn't make directly. "Any
            # non-deleted status can be suppressed", so without this a report could be laundered
            # through the archive into RESOLVED from a status the model refuses to resolve from —
            # candidate/in_progress (landing in RESOLVED with no title or summary), or failed
            # (a failed pipeline run presented as successfully resolved). So a suppressed report may
            # only resolve when it was in a directly-resolvable status before being archived.
            # Deliberately narrower than restore_target_status()'s researched set, which exists to
            # answer "where does restore put this report back", not "may it resolve".
            if (
                report.status == SignalReport.Status.SUPPRESSED
                and target_status == SignalReport.Status.RESOLVED
                and report.status_before_suppression not in _RESOLVABLE_STATUSES_BEFORE_SUPPRESSION
            ):
                return (
                    SignalReportBulkStateOutcome.SKIPPED,
                    "Only a report that was ready, awaiting input, or already resolved when it was "
                    "archived can be resolved from the archive.",
                )

            # "potential" on a suppressed report means "restore" (un-archive): return it to the state
            # it held before suppression when that was a researched, user-visible report, instead of
            # always dropping back to potential. snooze_for is irrelevant here and ignored.
            effective_target = target_status
            if report.status == SignalReport.Status.SUPPRESSED and target_status == SignalReport.Status.POTENTIAL:
                effective_target = report.restore_target_status()

            effective_snooze_for = snooze_for if target == "potential" else None

            try:
                updated_fields = report.transition_to(effective_target, snooze_for=effective_snooze_for)
            except InvalidStatusTransition as e:
                logger.warning("Invalid status transition for SignalReport %s: %s", report.id, e, exc_info=True)
                return SignalReportBulkStateOutcome.SKIPPED, None
            except (ValueError, TypeError) as e:
                logger.warning("Invalid data when transitioning SignalReport %s: %s", report.id, e, exc_info=True)
                return SignalReportBulkStateOutcome.FAILED, None

            writes_dismissal_feedback = target in ("suppressed", "potential", "resolved") and bool(
                dismissal_reason or dismissal_note
            )
            # Tell the status-changed receiver that this transition carries caller-supplied feedback,
            # so the label picks up the artefact written just below. Set before the save because the
            # post_save receiver snapshots it there. Matters most for resolve: a resolve driven by
            # the PR-merge webhook has no feedback, and only this flag distinguishes the two.
            report._wrote_dismissal_feedback = writes_dismissal_feedback  # type: ignore[attr-defined]

            report.save(update_fields=updated_fields)

            # Persist the dismissal feedback as its own artefact so it survives status changes
            # and so multiple dismissals (with different rationales) can stack over time.
            # Captured for suppress, snooze (transition to potential), and resolve flows — on a
            # resolve it records why the report was resolved with user attribution.
            if writes_dismissal_feedback:
                user = self.request.user
                is_authenticated = getattr(user, "is_authenticated", False)
                user_uuid = getattr(user, "uuid", None) if is_authenticated else None
                SignalReportArtefact.append_dismissal(
                    team_id=self.team.id,
                    report_id=str(report.id),
                    content=Dismissal(
                        reason=dismissal_reason,
                        note=dismissal_note,
                        user_id=getattr(user, "id", None) if is_authenticated else None,
                        user_uuid=str(user_uuid) if user_uuid else None,
                    ),
                    attribution=self._request_attribution(),
                )
                # The dismissal prefetch may have been evaluated before this artefact
                # existed; drop the stale cache so a follow-up serializer re-reads the
                # just-written reason/note instead of the previous (or empty) dismissal.
                if hasattr(report, "prefetched_dismissal_artefacts"):
                    del report.prefetched_dismissal_artefacts

        # A dismissal (transition into SUPPRESSED) closes the linked implementation PR — handled
        # centrally by the post_save receiver (receivers.close_pr_when_report_dismissed), so this
        # method doesn't special-case it. Restore/snooze to "potential" leaves the PR alone.
        return SignalReportBulkStateOutcome.TRANSITIONED, None

    @extend_schema(
        request=SignalReportBulkStateRequestSerializer,
        responses={200: SignalReportBulkStateResponseSerializer},
    )
    @action(detail=False, methods=["post"], url_path="bulk-state", required_scopes=["task:write"])
    def bulk_state(self, request, **kwargs):
        """
        Transition many reports to a new state in one call.

        Each id is processed independently: a report whose transition isn't allowed from its
        current status is reported as `skipped` (a 409 on the single-report endpoint) and the
        rest still go through. Returns one result per requested id (in request order, after
        de-duplication) plus per-outcome counts. The whole call is 200 even on partial failure —
        inspect `results` / the counts to see what happened.
        """
        serializer = SignalReportBulkStateRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        target = data["state"]
        dismissal_reason = data.get("dismissal_reason")
        dismissal_note = data.get("dismissal_note")
        snooze_for = data.get("snooze_for")

        # De-duplicate while preserving request order so the response lines up with what was asked.
        ordered_ids: list[str] = []
        seen: set[str] = set()
        for raw_id in data["ids"]:
            key = str(raw_id)
            if key not in seen:
                seen.add(key)
                ordered_ids.append(key)

        # One team-scoped fetch for every requested report. get_queryset() scopes to the team and
        # excludes deleted reports; suppressed reports are reachable because bulk_state is in
        # _SUPPRESSED_VISIBLE_ACTIONS, so bulk restore (state='potential') works too.
        reports_by_id = {str(report.id): report for report in self.get_queryset().filter(id__in=ordered_ids)}

        results: list[dict] = []
        counts: dict[str, int] = {outcome.value: 0 for outcome in SignalReportBulkStateOutcome}
        transitioned: list[SignalReport] = []
        for report_id in ordered_ids:
            report = reports_by_id.get(report_id)
            if report is None:
                outcome = SignalReportBulkStateOutcome.NOT_FOUND
                detail = None
                report_status = None
            else:
                outcome, detail = self._transition_report_state(
                    report,
                    target=target,
                    dismissal_reason=dismissal_reason,
                    dismissal_note=dismissal_note,
                    snooze_for=snooze_for,
                )
                report_status = report.status if outcome == SignalReportBulkStateOutcome.TRANSITIONED else None
                if outcome == SignalReportBulkStateOutcome.TRANSITIONED:
                    transitioned.append(report)
            results.append(
                {
                    "id": report_id,
                    "outcome": outcome.value,
                    "status": report_status,
                    "detail": detail or self._BULK_STATE_OUTCOME_DETAIL.get(outcome),
                }
            )
            counts[outcome.value] += 1

        self._forward_dismissal_note(
            reports=transitioned,
            dismissal_reason=dismissal_reason,
            dismissal_note=dismissal_note,
        )

        return Response(
            {
                "results": results,
                "transitioned_count": counts[SignalReportBulkStateOutcome.TRANSITIONED.value],
                "skipped_count": counts[SignalReportBulkStateOutcome.SKIPPED.value],
                "failed_count": counts[SignalReportBulkStateOutcome.FAILED.value],
                "not_found_count": counts[SignalReportBulkStateOutcome.NOT_FOUND.value],
            }
        )

    def _signals_pr_refunds_enabled(self) -> bool:
        # Server-side feature gate, keyed on the org (the refund window is the org's billing
        # period). Same call shape as the quota-limiting data-retention flag check.
        return (
            hanzo_insights.feature_enabled(
                SIGNALS_PR_REFUNDS_FEATURE_FLAG,
                str(self.organization.id),
                groups={"organization": str(self.organization.id)},
                group_properties={"organization": {"id": str(self.organization.id)}},
            )
            is True
        )

    def _refund_response(self, refund: SignalReportRefund, *, already_refunded: bool = False) -> Response:
        if already_refunded:
            # SignalReportRefundResponseSerializer reads the attribute; absent it falls back to
            # the field default (False), so only the idempotent-replay path sets it.
            refund.already_refunded = True  # type: ignore[attr-defined]
        return Response(SignalReportRefundResponseSerializer(refund).data)

    @validated_request(
        request_serializer=SignalReportRefundRequestSerializer,
        responses={
            200: OpenApiResponse(
                response=SignalReportRefundResponseSerializer,
                description=(
                    "The refund (created, or the existing one with already_refunded=true when the "
                    "report was refunded before)."
                ),
            ),
            400: OpenApiResponse(
                description=(
                    "Report is not refundable: no billable implementation PR, the PR run is outside "
                    "the current billing period, or the report is system-marked never-billable."
                )
            ),
            404: OpenApiResponse(description="Report not found, or refunds are not enabled for this organization."),
        },
        summary="Refund a report's implementation PR",
        description=(
            "Refund the flat charge for this report's implementation PR and archive the report. "
            "Refunds auto-approve: the charge is either excluded from usage before it is ever "
            "reported to billing (refund on the same UTC day as the PR run) or returned as a Stripe "
            "customer-balance credit on the next invoice. A refunded PR does not count toward the "
            "free monthly PR allowance. One refund per report, ever — repeat calls return the "
            "existing refund with already_refunded=true. The report is archived as part of the "
            "refund (a resolved report stays resolved) and can't be restored afterwards."
        ),
        operation_id="signals_reports_refund_create",
    )
    @action(detail=True, methods=["post"], url_path="refund", required_scopes=["task:write"])
    def refund(self, request: ValidatedRequest, pk=None, **kwargs):
        if not self._signals_pr_refunds_enabled():
            raise NotFound("PR refunds are not enabled for this organization.")

        report = cast(SignalReport, self.get_object())
        data = request.validated_data
        note = data.get("note") or ""
        # Resolved before the locked transaction — a bad X-Insights-Task-Id header must 400
        # before anything mutates (mirrors the artefact write paths).
        attribution = resolve_request_attribution(request, self.team.id)
        user = cast(User, request.user)

        with transaction.atomic():
            # Row-lock the report so double-clicks serialize (precedent: the Slack suppression
            # flow); the refund OneToOne is the backstop if something bypasses the lock.
            report = cast(
                SignalReport,
                SignalReport.objects.select_for_update().get(id=report.id, team_id=self.team_id),
            )

            existing = getattr(report, "refund", None)
            if existing is not None:
                return self._refund_response(existing, already_refunded=True)

            # Same decision the serializer's `refund_ineligibility_reason` field renders from,
            # so the UI's button state and this endpoint's 400s can never disagree.
            billable_run = first_billable_pr_run(report.id)
            # Computed once and frozen onto the refund row below: the credited-path sync must
            # report the period the refund was accepted in, not whatever period is current when
            # the Celery task eventually reaches billing.
            period = current_billing_period_bounds(self.organization)
            ineligibility = refund_ineligibility_reason(
                has_refund=False,  # the idempotent 200 above already handled existing refunds
                billing_exempt=bool(report.billing_exempt_reason),
                billable_run_at=billable_run.created_at if billable_run else None,
                period=period,
            )
            if ineligibility is not None:
                return Response(
                    {"error": _REFUND_INELIGIBLE_MESSAGES[ineligibility]},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            assert billable_run is not None  # narrowed by the no_billable_pr reason above

            # Freeze the billing path by the UTC-day rule: refunds committed on the PR run's own
            # UTC day land before any usage send covering that day (all sends happen ≥3h45m into
            # the next UTC day), so the report can simply be excluded from usage; anything later
            # must go through a billing-service credit. Decided once, stored, never recomputed.
            now = timezone.now()
            # Derived from the persisted merge flag, not the report status: a report can be resolved
            # manually without a merged PR, so RESOLVED no longer implies the PR shipped. Asked about
            # the PR being refunded specifically — it's that PR the refund reverses the charge for,
            # and that PR which has to be closed if it never merged.
            pr_merged = report_pr_is_merged(report.id, billable_run.pr_url)
            billing_path = (
                SignalReportRefund.BillingPath.EXCLUDED
                if billable_run.created_at.astimezone(UTC).date() == now.astimezone(UTC).date()
                else SignalReportRefund.BillingPath.CREDITED
            )

            try:
                # Savepoint so a unique-constraint loser doesn't poison the outer transaction.
                with transaction.atomic():
                    refund = SignalReportRefund.objects.create(
                        team_id=report.team_id,
                        report=report,
                        created_by=user if user.is_authenticated else None,
                        reason=data["reason"],
                        note=note,
                        billing_path=billing_path,
                        credits=SIGNALS_CREDITS_PER_REPORT_WITH_PR,
                        pr_url=billable_run.pr_url,
                        pr_run_created_at=billable_run.created_at,
                        period_start=period.start,
                        period_end=period.end,
                    )
            except IntegrityError:
                existing = SignalReportRefund.objects.filter(report_id=report.id).first()
                if existing is None:
                    raise
                return self._refund_response(existing, already_refunded=True)

            # Refund doubles as archive: suppress the report so it leaves the inbox, and so the
            # dismissal receiver closes the implementation PR that was paid for and then refunded.
            # The one exception is a report resolved by a merged PR: that PR shipped, so the report
            # stays put as a genuine terminal state and there is no open PR to close. A report
            # resolved manually without a merged PR is NOT exempt — its PR may still be open, and
            # leaving it resolved would let the caller keep the implementation work after the refund.
            # Already-suppressed reports need no transition. The dismissal artefact records the
            # rationale; the structured truth lives on the refund row.
            resolved_via_merged_pr = report.status == SignalReport.Status.RESOLVED and pr_merged
            if report.status != SignalReport.Status.SUPPRESSED and not resolved_via_merged_pr:
                updated_fields = report.transition_to(SignalReport.Status.SUPPRESSED)
                report.save(update_fields=updated_fields)
            SignalReportArtefact.append_dismissal(
                team_id=self.team.id,
                report_id=str(report.id),
                content=Dismissal(
                    reason="refunded",
                    note=note or None,
                    user_id=user.id if user.is_authenticated else None,
                    user_uuid=str(user.uuid) if user.is_authenticated else None,
                ),
                attribution=attribution,
            )

            if billing_path == SignalReportRefund.BillingPath.CREDITED:
                refund_id = str(refund.id)
                transaction.on_commit(lambda: sync_signals_refund_credit.delay(refund_id))

        report_user_action(
            user,
            "signals_pr_refund_created",
            properties={
                "team_id": self.team.id,
                "organization_id": str(self.organization.id),
                "report_id": str(report.id),
                "refund_id": str(refund.id),
                "billing_path": billing_path,
                "credits": refund.credits,
                "reason": refund.reason,
                "pr_merged": pr_merged,
                "days_since_pr": (now - billable_run.created_at).days,
            },
            team=self.team,
            organization=self.organization,
        )

        return self._refund_response(refund)

    @extend_schema(
        responses={
            200: OpenApiResponse(
                response=SignalReportRefundSummaryResponseSerializer,
                description="Org-wide credited-refund totals for the current billing period.",
            ),
            404: OpenApiResponse(description="Refunds are not enabled for this organization."),
        },
        summary="Summarize credited PR refunds for the billing period",
        description=(
            "Aggregate credited-path refunds across the whole organization for the current billing "
            "period — counts only, no per-team detail. The billing usage widget needs this because "
            "billing usage is org-wide while reports (and their refunds) are team-scoped: subtract "
            "the refunded credits from billing usage to show the net PR count. Excluded-path "
            "refunds never reach billing usage, so no adjustment is needed for them. Also carries "
            "the org's live billable credits for the period (billing's recorded usage lags by up "
            "to a day), so the widget can count just-created PRs and react to same-day refunds."
        ),
        operation_id="signals_reports_refund_summary_retrieve",
    )
    @action(detail=False, methods=["get"], url_path="refund-summary", required_scopes=["task:read"])
    def refund_summary(self, request: Request, **kwargs):
        # Two independent flags can each make this endpoint matter: refunds (the netting numbers)
        # and quota enforcement (`quota_limited` drives the widget's paused banner, and enforcement
        # can roll out before refunds). Gating on refunds alone would hide the paused state for
        # enforcement-only orgs.
        if not (self._signals_pr_refunds_enabled() or self_driving_quota_enforcement_enabled(self.team)):
            raise NotFound("PR refunds are not enabled for this organization.")

        period = current_billing_period_bounds(self.organization)
        aggregates = (
            # Org-wide on purpose (unscoped + org filter): the usage this offsets is org-level.
            SignalReportRefund.objects.unscoped()
            .filter(
                team__organization_id=self.organization.id,
                billing_path=SignalReportRefund.BillingPath.CREDITED,
                pr_run_created_at__gte=period.start,
                pr_run_created_at__lt=period.end,
            )
            .aggregate(credited_refund_count=Count("id"), credited_credits=Sum("credits"))
        )
        return Response(
            {
                "credited_refund_count": aggregates["credited_refund_count"] or 0,
                "credited_credits": aggregates["credited_credits"] or 0,
                "period_billable_credits": period_billable_credits_for_org(self.organization.id, period=period),
                "quota_limited": self_driving_quota_gate(self.team).enforced,
            }
        )

    @extend_schema(exclude=True)
    @action(detail=True, methods=["post"], url_path="reingest", required_scopes=["task:write"])
    def reingest(self, request, pk=None, **kwargs):
        """Re-ingest a report's signals (same team access as other report actions)."""
        report = cast(SignalReport, self.get_object())
        report_id = str(report.id)
        team_id = self.team.id

        try:
            client = sync_connect()
            async_to_sync(client.start_workflow)(  # type: ignore
                "signal-report-reingestion",  # type: ignore
                SignalReportReingestionWorkflowInputs(team_id=team_id, report_id=report_id),  # type: ignore
                id=SignalReportReingestionWorkflow.workflow_id_for(team_id, report_id),
                task_queue=settings.VIDEO_EXPORT_TASK_QUEUE,
                execution_timeout=timedelta(minutes=30),
                retry_policy=RetryPolicy(maximum_attempts=1),
            )
        except WorkflowAlreadyStartedError:
            return Response({"status": "already_running", "report_id": report_id}, status=status.HTTP_200_OK)
        except Exception:
            logger.exception("Failed to start reingestion workflow for report %s", report_id)
            return Response(
                {"error": "Failed to start reingestion workflow."},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )

        return Response({"status": "reingestion_started", "report_id": report_id}, status=status.HTTP_202_ACCEPTED)

    def _resolve_report_pr_reference(self, report: SignalReport) -> tuple[str, int] | None:
        """Resolve a report's implementation PR to ``(owner/repo, pr_number)``, or None if it has none
        (or the stored URL isn't a parseable GitHub PR URL)."""
        pr_url = fetch_implementation_pr_urls_for_reports([str(report.id)]).get(str(report.id))
        if not pr_url:
            return None
        parsed = GitHubIntegration.parse_pull_request_url(pr_url)
        if parsed is None:
            return None
        return parsed.repository, parsed.number

    @extend_schema(
        responses={
            200: OpenApiResponse(
                response=PullRequestChecksResponseSerializer,
                description="The CI checks on the report's implementation pull request.",
            ),
            404: OpenApiResponse(
                description="Report has no implementation PR, or no GitHub integration can access it."
            ),
            502: OpenApiResponse(description="GitHub could not return the checks."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Fetch CI checks for a report's implementation PR",
        description=(
            "Fetch the CI status (GitHub Actions check runs and legacy commit statuses) of the pull "
            "request the report's implementation task opened, via the team's GitHub integration."
        ),
        operation_id="signals_report_pr_checks",
    )
    @action(detail=True, methods=["get"], url_path="pr_checks", required_scopes=["task:read"])
    def pr_checks(self, request: Request, *args, **kwargs) -> Response:
        return self._pr_github_passthrough(
            cast(SignalReport, self.get_object()), "get_pull_request_checks", "checks", "checks"
        )

    @extend_schema(
        responses={
            200: OpenApiResponse(
                response=PullRequestCommentsResponseSerializer,
                description="Conversation and review comments on the report's implementation pull request.",
            ),
            404: OpenApiResponse(
                description="Report has no implementation PR, or no GitHub integration can access it."
            ),
            502: OpenApiResponse(description="GitHub could not return the comments."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Fetch comments for a report's implementation PR",
        description=(
            "Fetch the pull request's conversation comments and inline review comments, merged "
            "chronologically, via the team's GitHub integration."
        ),
        operation_id="signals_report_pr_comments",
    )
    @action(detail=True, methods=["get"], url_path="pr_comments", required_scopes=["task:read"])
    def pr_comments(self, request: Request, *args, **kwargs) -> Response:
        return self._pr_github_passthrough(
            cast(SignalReport, self.get_object()), "get_pull_request_comments", "comments", "comments"
        )

    @extend_schema(
        request=PullRequestReviewCommentCreateSerializer,
        responses={
            201: OpenApiResponse(
                response=PullRequestReviewCommentCreateResponseSerializer,
                description="The created review comment, in the normalized PR-comment shape.",
            ),
            400: OpenApiResponse(description="Invalid comment payload."),
            403: OpenApiResponse(
                description="Not a signed-in user, or the requesting user has no usable personal GitHub "
                "connection (reconnect GitHub)."
            ),
            404: OpenApiResponse(
                description="Report has no implementation PR, or no GitHub integration can access it."
            ),
            502: OpenApiResponse(description="GitHub rejected the comment."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Post an inline review comment on a report's implementation PR",
        description=(
            "Post an inline review comment on the report's implementation pull request, attributed to the "
            "requesting user's own GitHub identity via their personal GitHub connection. Either replies to "
            "an existing thread (`in_reply_to`) or starts a new thread on a diff line (`path` + `line`)."
        ),
        operation_id="signals_report_pr_review_comments_create",
    )
    @action(detail=True, methods=["post"], url_path="pr_review_comments", required_scopes=["task:write"])
    def pr_review_comments(self, request: Request, *args, **kwargs) -> Response:
        report = cast(SignalReport, self.get_object())
        serializer = PullRequestReviewCommentCreateSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        params = serializer.validated_data

        resolved = self._resolve_user_github_and_pr(report, request)
        if isinstance(resolved, Response):
            return resolved
        user_github, repository, pr_number = resolved

        commit_id: str | None = None
        if not params.get("in_reply_to"):
            # New threads anchor to the PR head commit; resolve it via the team integration
            # (read path), so the user token is only used for the write itself.
            github, repository, pr_number, error = self._github_for_report_pr(report, reference=(repository, pr_number))
            if error is not None:
                return error
            assert github is not None
            pr_details = github.get_pull_request(repository, pr_number)
            commit_id = pr_details.get("head_sha") if pr_details.get("success") else None
            if not commit_id:
                return Response(
                    {"error": "Could not resolve the pull request's head commit."},
                    status=status.HTTP_502_BAD_GATEWAY,
                )

        result = self._run_user_github_write(
            lambda: user_github.create_pull_request_review_comment(
                repository,
                pr_number,
                params["body"],
                in_reply_to=params.get("in_reply_to"),
                commit_id=commit_id,
                path=params.get("path"),
                line=params.get("line"),
                side=params.get("side"),
            ),
            repository,
            pr_number,
            noun="comment",
        )
        if isinstance(result, Response):
            return result

        self._bust_pr_comments_cache(repository, pr_number)
        return self._pr_review_comment_response(result["comment"], status_code=status.HTTP_201_CREATED)

    @extend_schema(
        methods=["PATCH"],
        parameters=[OpenApiParameter("comment_id", OpenApiTypes.STR, OpenApiParameter.PATH)],
        request=PullRequestReviewCommentUpdateSerializer,
        responses={
            200: OpenApiResponse(
                response=PullRequestReviewCommentCreateResponseSerializer,
                description="The edited review comment, in the normalized PR-comment shape.",
            ),
            403: OpenApiResponse(
                description="Not a signed-in user, no usable personal GitHub connection, or not the comment's author."
            ),
            404: OpenApiResponse(description="Report has no implementation PR, or the comment isn't on it."),
            502: OpenApiResponse(description="GitHub rejected the edit."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Edit one of the requesting user's own review comments",
        operation_id="signals_report_pr_review_comment_update",
    )
    @extend_schema(
        methods=["DELETE"],
        parameters=[OpenApiParameter("comment_id", OpenApiTypes.STR, OpenApiParameter.PATH)],
        responses={
            204: OpenApiResponse(description="Comment deleted."),
            403: OpenApiResponse(
                description="Not a signed-in user, no usable personal GitHub connection, or not the comment's author."
            ),
            404: OpenApiResponse(description="Report has no implementation PR, or the comment isn't on it."),
            502: OpenApiResponse(description="GitHub rejected the delete."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Delete one of the requesting user's own review comments",
        operation_id="signals_report_pr_review_comment_destroy",
    )
    @action(
        detail=True,
        methods=["patch", "delete"],
        url_path=r"pr_review_comments/(?P<comment_id>[0-9]+)",
        required_scopes=["task:write"],
    )
    def pr_review_comment(self, request: Request, *args, **kwargs) -> Response:
        report = cast(SignalReport, self.get_object())
        comment_id = str(kwargs["comment_id"])
        resolved = self._resolve_user_github_and_pr(report, request, comment_id=comment_id)
        if isinstance(resolved, Response):
            return resolved
        user_github, repository, pr_number = resolved

        def run() -> dict[str, Any]:
            if request.method == "DELETE":
                return user_github.delete_pull_request_review_comment(repository, comment_id)
            serializer = PullRequestReviewCommentUpdateSerializer(data=request.data)
            serializer.is_valid(raise_exception=True)
            return user_github.update_pull_request_review_comment(
                repository, comment_id, serializer.validated_data["body"]
            )

        result = self._run_user_github_write(run, repository, pr_number, noun="comment")
        if isinstance(result, Response):
            return result

        self._bust_pr_comments_cache(repository, pr_number)
        if request.method == "DELETE":
            return Response(status=status.HTTP_204_NO_CONTENT)
        return self._pr_review_comment_response(result["comment"], status_code=status.HTTP_200_OK)

    @extend_schema(
        request=PullRequestReviewCommentReactionCreateSerializer,
        responses={
            201: OpenApiResponse(
                response=PullRequestReviewCommentReactionCreateResponseSerializer,
                description="The created reaction.",
            ),
            403: OpenApiResponse(description="Not a signed-in user, or no usable personal GitHub connection."),
            404: OpenApiResponse(description="Report has no implementation PR, or the comment isn't on it."),
            502: OpenApiResponse(description="GitHub rejected the reaction."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="React to a review comment as the requesting user",
        operation_id="signals_report_pr_review_comment_reactions_create",
        parameters=[OpenApiParameter("comment_id", OpenApiTypes.STR, OpenApiParameter.PATH)],
    )
    @action(
        detail=True,
        methods=["post"],
        url_path=r"pr_review_comments/(?P<comment_id>[0-9]+)/reactions",
        required_scopes=["task:write"],
    )
    def pr_review_comment_reactions(self, request: Request, *args, **kwargs) -> Response:
        report = cast(SignalReport, self.get_object())
        comment_id = str(kwargs["comment_id"])
        serializer = PullRequestReviewCommentReactionCreateSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        resolved = self._resolve_user_github_and_pr(report, request, comment_id=comment_id)
        if isinstance(resolved, Response):
            return resolved
        user_github, repository, pr_number = resolved

        result = self._run_user_github_write(
            lambda: user_github.add_pull_request_review_comment_reaction(
                repository, comment_id, serializer.validated_data["content"]
            ),
            repository,
            pr_number,
            noun="reaction",
        )
        if isinstance(result, Response):
            return result

        self._bust_pr_comments_cache(repository, pr_number)
        raw = result.get("reaction") or {}
        reaction = {
            "id": str(raw["id"]) if raw.get("id") is not None else "",
            "content": raw.get("content"),
            "user_login": (raw.get("user") or {}).get("login"),
        }
        return Response({"reaction": reaction}, status=status.HTTP_201_CREATED)

    @extend_schema(
        responses={
            204: OpenApiResponse(description="Reaction removed."),
            403: OpenApiResponse(description="Not a signed-in user, or no usable personal GitHub connection."),
            404: OpenApiResponse(description="Report has no implementation PR, or the comment isn't on it."),
            502: OpenApiResponse(description="GitHub rejected the removal."),
            503: OpenApiResponse(description="The GitHub egress budget is temporarily unavailable."),
        },
        summary="Remove one of the requesting user's own reactions from a review comment",
        operation_id="signals_report_pr_review_comment_reaction_destroy",
        parameters=[
            OpenApiParameter("comment_id", OpenApiTypes.STR, OpenApiParameter.PATH),
            OpenApiParameter("reaction_id", OpenApiTypes.STR, OpenApiParameter.PATH),
        ],
    )
    @action(
        detail=True,
        methods=["delete"],
        url_path=r"pr_review_comments/(?P<comment_id>[0-9]+)/reactions/(?P<reaction_id>[0-9]+)",
        required_scopes=["task:write"],
    )
    def pr_review_comment_reaction(self, request: Request, *args, **kwargs) -> Response:
        report = cast(SignalReport, self.get_object())
        comment_id = str(kwargs["comment_id"])
        reaction_id = str(kwargs["reaction_id"])
        resolved = self._resolve_user_github_and_pr(report, request, comment_id=comment_id)
        if isinstance(resolved, Response):
            return resolved
        user_github, repository, pr_number = resolved

        result = self._run_user_github_write(
            lambda: user_github.delete_pull_request_review_comment_reaction(repository, comment_id, reaction_id),
            repository,
            pr_number,
            noun="reaction",
        )
        if isinstance(result, Response):
            return result
        self._bust_pr_comments_cache(repository, pr_number)
        return Response(status=status.HTTP_204_NO_CONTENT)

    def _bust_pr_comments_cache(self, repository: str, pr_number: int) -> None:
        """Drop the short PR-comments read cache so a just-made change shows on the next fetch."""
        cache.delete(f"signals:pr-github:{self.team.id}:{repository}:{pr_number}:get_pull_request_comments")

    @staticmethod
    def _pr_review_comment_response(raw: object, *, status_code: int) -> Response:
        """Shape a raw GitHub review comment via the shared read-path normalizer (a fresh write never
        carries reactions). The shared transport treats a body-less success as valid, so a write result
        we can't normalize is an upstream failure rather than anything the caller did wrong."""
        normalized = GitHubIntegrationBase.normalize_pr_comment(raw, "review")
        if normalized is None:
            return Response(
                {"error": "GitHub returned the comment in an unexpected format. Reload to see the latest."},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        return Response({"comment": normalized}, status=status_code)

    def _resolve_user_github_and_pr(
        self, report: SignalReport, request: Request, *, comment_id: str | None = None
    ) -> tuple[UserGitHubIntegration, str, int] | Response:
        """Resolve the requesting user's personal GitHub integration and the report's PR, or a Response
        error (404 no PR, 403 not a human caller or no personal GitHub connection) to return as-is.

        Every write that reaches GitHub under a human's own identity funnels through here, so the
        human-caller check lives here too rather than on each action. It admits only a browser
        session instead of rejecting known token types, because every credential that authenticates
        *as* the user would otherwise inherit their linked GitHub account: a sandbox agent token is
        minted as the task actor and carries ``task:write``, and a personal API key holding the same
        scope is issued for automation rather than for acting as that person on GitHub. Either one
        could comment, edit, delete, or merge as them.

        ``comment_id``, when given, is resolved and required to sit on the report's own PR. GitHub's
        review-comment edit, delete, and reaction endpoints are repository-wide, so this is the only
        thing keeping a report from reaching comments on every other PR in its repository.
        """
        if not isinstance(request.successful_authenticator, SessionAuthentication):
            return Response(
                {"error": "This must be done by a signed-in user, not an automated token."},
                status=status.HTTP_403_FORBIDDEN,
            )
        user = cast(User, request.user)
        reference = self._resolve_report_pr_reference(report)
        if reference is None:
            return Response(
                {"error": "This report has no implementation pull request."}, status=status.HTTP_404_NOT_FOUND
            )
        user_integration = (
            UserIntegration.objects.filter(user=user, kind=UserIntegration.IntegrationKind.GITHUB)
            .order_by("created_at")
            .first()
        )
        if user_integration is None:
            return Response(
                {"error": "Connect your GitHub account in settings to comment on pull requests."},
                status=status.HTTP_403_FORBIDDEN,
            )
        repository, pr_number = reference
        user_github = UserGitHubIntegration(user_integration)
        if comment_id is not None:
            lookup = self._run_user_github_write(
                lambda: user_github.get_pull_request_review_comment(repository, comment_id),
                repository,
                pr_number,
                noun="comment",
            )
            if isinstance(lookup, Response):
                return lookup
            # Matched on the suffix, not the whole URL: the repository is already pinned by the path we
            # just fetched, while the report's stored PR URL can differ from GitHub's canonical casing.
            pull_request_url = (lookup.get("comment") or {}).get("pull_request_url")
            if not isinstance(pull_request_url, str) or not pull_request_url.endswith(f"/pulls/{pr_number}"):
                return Response(
                    {"error": "This comment isn't on this report's pull request. Refresh the page and try again."},
                    status=status.HTTP_404_NOT_FOUND,
                )
        return user_github, repository, pr_number

    def _run_user_github_write(
        self, fn: Callable[[], dict[str, Any]], repository: str, pr_number: int, *, noun: str
    ) -> dict[str, Any] | Response:
        """Run a user-authored GitHub write, mapping the shared failure modes to Responses. Returns the
        call's result dict on success, or a Response to return as-is. GitHub 403 (e.g. editing someone
        else's comment) surfaces as a 403."""
        try:
            result = fn()
        except ReauthorizationRequired:
            return Response(
                {"error": "Your GitHub connection has expired. Reconnect GitHub in settings."},
                status=status.HTTP_403_FORBIDDEN,
            )
        except GitHubRateLimitError as e:
            return github_rate_limited_response(e)
        except GitHubEgressBudgetExhausted:
            return Response(
                {"error": "GitHub is temporarily busy. Try again shortly."},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )
        except Exception:  # noqa: BLE001 — never let an upstream GitHub failure 500 this endpoint
            logger.warning(f"signals pr review {noun} write errored", repository=repository, pr_number=pr_number)
            return Response({"error": f"GitHub could not accept the {noun}."}, status=status.HTTP_502_BAD_GATEWAY)

        if not result.get("success"):
            forbidden = result.get("status_code") == 403
            return Response(
                {"error": result.get("error") or f"GitHub could not accept the {noun}."},
                status=status.HTTP_403_FORBIDDEN if forbidden else status.HTTP_502_BAD_GATEWAY,
            )
        return result

    def _pr_github_passthrough(self, report: SignalReport, fetch_name: str, key: str, noun: str) -> Response:
        """Shared body for the ``pr_checks`` / ``pr_comments`` actions: resolve the report's PR and the
        GitHub integration that can read it, call ``fetch_name`` on the client, and return ``{key: ...}``.
        A missing PR/integration maps to 404, a rate limit to its response, and any other upstream GitHub
        failure to 502 — an upstream hiccup never 500s the endpoint."""
        reference = self._resolve_report_pr_reference(report)
        if reference is None:
            return Response(
                {"error": "This report has no implementation pull request."},
                status=status.HTTP_404_NOT_FOUND,
            )
        repository, pr_number = reference
        cache_key = f"signals:pr-github:{self.team.id}:{repository}:{pr_number}:{fetch_name}"
        cached_result = cache.get(cache_key)
        if isinstance(cached_result, dict) and key in cached_result:
            return Response({key: cached_result[key]})

        github, repository, pr_number, error = self._github_for_report_pr(report, reference=reference)
        if error is not None:
            return error
        assert github is not None  # `error is None` guarantees a resolved integration
        try:
            result: dict[str, Any] = getattr(github, fetch_name)(repository, pr_number)
        except GitHubRateLimitError as e:
            return github_rate_limited_response(e)
        except GitHubEgressBudgetExhausted:
            return Response(
                {"error": "GitHub is temporarily busy. Try again shortly."},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )
        except Exception:  # noqa: BLE001 — never let an upstream GitHub failure 500 this endpoint
            logger.warning(f"signals pr {noun} fetch errored", repository=repository, pr_number=pr_number)
            return Response(
                {"error": f"GitHub could not return the {noun} for this pull request."},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        if not result.get("success"):
            return Response(
                {"error": result.get("error") or f"GitHub could not return the {noun} for this pull request."},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        cache.set(cache_key, result, timeout=PR_GITHUB_CACHE_SECONDS)
        return Response({key: result[key]})

    def _github_for_report_pr(
        self,
        report: SignalReport,
        *,
        reference: tuple[str, int] | None = None,
    ) -> tuple[GitHubIntegration | None, str, int, Response | None]:
        """Resolve the report's implementation PR and the GitHub integration that can read it.

        Returns ``(github, repository, pr_number, error_response)`` — on any failure ``error_response``
        is set (and should be returned as-is) and ``github`` is None. Mirrors the connection-boundary
        scoping of the artefact `diff` action: access is bounded to repos the team's installation can
        reach, not to a single per-report repository.
        """
        reference = reference or self._resolve_report_pr_reference(report)
        if reference is None:
            return (
                None,
                "",
                0,
                Response(
                    {"error": "This report has no implementation pull request."},
                    status=status.HTTP_404_NOT_FOUND,
                ),
            )
        repository, pr_number = reference
        try:
            github = GitHubIntegration.first_for_team_repository(
                self.team.id,
                repository,
                source="signals_pr_detail",
                priority=Priority.NORMAL,
            )
        except GitHubRateLimitError as e:
            return None, repository, pr_number, github_rate_limited_response(e)
        except GitHubEgressBudgetExhausted:
            return (
                None,
                repository,
                pr_number,
                Response(
                    {"error": "GitHub is temporarily busy. Try again shortly."},
                    status=status.HTTP_503_SERVICE_UNAVAILABLE,
                ),
            )
        if github is None:
            return (
                None,
                repository,
                pr_number,
                Response(
                    {"error": f"No GitHub integration can access '{repository}'."},
                    status=status.HTTP_404_NOT_FOUND,
                ),
            )
        return github, repository, pr_number, None


# `report_id` addresses a report's UUID primary key. Agents whose prompt only carries a
# `signal_id` sometimes pass that (e.g. `sig_praise`) here instead, so make the constraint
# explicit in the schema — a non-UUID value can never match a report.
_REPORT_ID_PARAMETER = OpenApiParameter(
    name="report_id",
    type=OpenApiTypes.UUID,
    location=OpenApiParameter.PATH,
    description=(
        "UUID of the report whose artefacts you're addressing. This must be a report id (the "
        "report's own UUID), not a signal id such as `sig_praise` — a non-report id returns 404."
    ),
)


class ReviewerWriteError(Exception):
    """A reviewer write payload that couldn't be resolved. Callers surface it as a 400 `{"error": ...}`."""


def _schedule_reviewer_added_slack_notifications(
    *, team_id: int, report_id: str, added_logins: Sequence[str], actor_user_id: int | None
) -> None:
    """After commit, Slack-notify reviewers a human just added to this report.

    Enqueued on commit so nothing is sent if the write rolls back, and so delivery's
    network calls (metadata lookups, Slack) run on a worker instead of holding up the
    request. Best-effort — a Slack failure must never break the reviewer edit.
    """
    # robust=True: a broker outage while enqueuing must not 500 an edit that already committed.
    transaction.on_commit(
        lambda: send_reviewer_added_slack_notifications.delay(
            report_id=report_id,
            team_id=team_id,
            added_github_logins=list(added_logins),
            exclude_user_id=actor_user_id,
        ),
        robust=True,
    )


def append_suggested_reviewers(
    *,
    team: Team,
    report_id: str,
    entries: list[dict],
    request: Request,
) -> tuple[SignalReportArtefact, set[str]]:
    """Append a new `suggested_reviewers` status row for a report, merging forward from the current
    (latest) reviewers. Works whether or not the report already has a reviewers artefact — the first
    write for a report with none simply creates the first row. Shared by the app-only reviewers PUT
    on both the report and artefact viewsets. Returns the new artefact and the set of canonical logins
    written (for read-time enrichment). Raises `ReviewerWriteError` on unresolvable entries."""
    # App/user-only write (agents append reviewers via the artefacts POST), so always attribute to
    # the requesting user — never the X-Insights-Task-Id header. A task-attributed reviewers row has
    # no created_by_id, which makes auto-start treat the list as agent-authored and run the
    # implementation task as a named colleague instead of the editor — the reviewer-impersonation
    # path the `triggering_user_id` guard in `auto_start` exists to close.
    user_id = request.user.id
    if user_id is None:  # unreachable behind authentication, but keeps attribution honest
        raise serializers.ValidationError("Cannot attribute a reviewer edit to an anonymous user.")
    attribution = ArtefactAttribution.from_user(user_id)

    # Resolve any user_uuid → canonical github_login via team org membership.
    uuids_to_resolve = [str(e["user_uuid"]) for e in entries if e.get("user_uuid")]
    uuid_to_login: dict[str, str] = (
        get_org_member_github_logins_by_user_uuid(team.id, uuids_to_resolve) if uuids_to_resolve else {}
    )

    # Resolve canonical login per entry. Fail loudly if a user_uuid does not
    # map to an org member with a GitHub identity on this team.
    # The bool tuple elements distinguish "github_name / reason explicitly supplied
    # (incl. empty string to clear)" from "field absent" — the merge step below
    # only falls back to the prior value when the field is absent.
    resolved_entries: list[tuple[str, str | None, bool, str | None, bool]] = []
    for idx, entry in enumerate(entries):
        user_uuid = entry.get("user_uuid")
        if user_uuid is not None:
            resolved_login = uuid_to_login.get(str(user_uuid))
            if not resolved_login:
                raise ReviewerWriteError(
                    f"content[{idx}]: user_uuid '{user_uuid}' is not an org member of this team "
                    "with a linked GitHub identity."
                )
            login_lc = resolved_login.lower()
        else:
            raw_login = entry.get("github_login") or ""
            login_lc = raw_login.strip().lower()
            if not login_lc:
                raise ReviewerWriteError(f"content[{idx}]: github_login resolved to empty after normalization.")

        explicit_name = "github_name" in entry
        github_name = entry.get("github_name") if explicit_name else None
        explicit_reason = "reason" in entry
        reason = entry.get("reason") if explicit_reason else None
        resolved_entries.append((login_lc, github_name, explicit_name, reason, explicit_reason))

    # Lock the report for the read-merge-append so concurrent reviewer edits serialize — each
    # write reads the current (latest) reviewers and appends a new row, so without the lock two
    # simultaneous edits would both read the same row and one would be silently lost.
    seen: set[str] = set()
    with transaction.atomic():
        report = (
            SignalReport.objects.select_for_update()
            .filter(id=report_id, team_id=team.id)
            .exclude(status=SignalReport.Status.DELETED)
            .first()
        )
        if report is None:
            # The report was concurrently soft-deleted between the caller's fetch and this lock;
            # don't append a stale reviewers row to a dead report.
            raise NotFound()

        # Merge commits/names forward from the *current* reviewers (the latest status row).
        # `suggested_reviewers` is append-only and latest-wins.
        current = (
            SignalReportArtefact.objects.filter(
                report_id=report_id,
                type=SignalReportArtefact.ArtefactType.SUGGESTED_REVIEWERS,
            )
            .order_by("-created_at")
            .first()
        )
        try:
            prior_content = json.loads(current.content) if current else []
        except (json.JSONDecodeError, ValueError):
            prior_content = []
        prior_commits_by_login: dict[str, list] = {}
        prior_name_by_login: dict[str, str | None] = {}
        prior_reason_by_login: dict[str, str | None] = {}
        prior_logins: list[str] = []
        if isinstance(prior_content, list):
            for prior in prior_content:
                if not isinstance(prior, dict):
                    continue
                login = (prior.get("github_login") or "").strip().lower()
                if not login:
                    continue
                prior_logins.append(login)
                commits = prior.get("relevant_commits")
                if isinstance(commits, list):
                    prior_commits_by_login[login] = commits
                prior_name = prior.get("github_name")
                if isinstance(prior_name, str):
                    prior_name_by_login[login] = prior_name
                prior_reason = prior.get("reason")
                if isinstance(prior_reason, str):
                    prior_reason_by_login[login] = prior_reason

        # Newly-added reviewers carry no routing evidence, so record who added them and when
        # (this path is always attributed to request.user). Dates use the report's project timezone.
        actor = cast(User, request.user)
        # Build the date without the platform-specific %-d directive (fails on non-Unix).
        now_local = timezone.now().astimezone(team.timezone_info)
        added_on = f"{now_local:%b} {now_local.day}, {now_local.year}"
        manual_add_reason = f"Added as a reviewer by {actor.get_full_name().strip() or actor.email} on {added_on}"

        # Dedupe by canonical login, preserve first-seen order.
        new_content: list[dict] = []
        for login_lc, github_name, explicit_name, reason, explicit_reason in resolved_entries:
            if login_lc in seen:
                continue
            seen.add(login_lc)
            # If the client supplied github_name (incl. ""), honour it. Otherwise
            # carry over the prior one so kept reviewers don't lose their name.
            # Same rule for reason. Only fall back to the manual-add note when the field was
            # omitted for a brand-new reviewer — an explicit null clears the reason, as for kept ones.
            effective_name = github_name if explicit_name else prior_name_by_login.get(login_lc)
            effective_reason = reason if explicit_reason else prior_reason_by_login.get(login_lc)
            if not explicit_reason and login_lc not in prior_logins:
                effective_reason = manual_add_reason
            new_content.append(
                {
                    "github_login": login_lc,
                    "github_name": effective_name,
                    "relevant_commits": prior_commits_by_login.get(login_lc, []),
                    "reason": effective_reason or None,
                }
            )

        # Append a new status row rather than mutating in place: a human reviewer edit becomes a
        # point-in-time entry in the work log, and latest-wins keeps it current. Appending a
        # reviewers status also re-evaluates auto-start (handled in `append_status`, on commit).
        new_artefact = SignalReportArtefact.append_status(
            team_id=team.id,
            report_id=str(report_id),
            content=SuggestedReviewers.model_validate(new_content),
            attribution=attribution,
        )

        # Human reviewer corrections are a routing signal (scouts query them via the
        # activity log to learn who owns an area), so log them — but only genuine
        # membership changes by a human, not agent writes or order-only rewrites.
        # `new_content` is deduped above; dedupe `prior_logins` too (a legacy or
        # hand-crafted prior row may carry duplicates) so before/after read symmetrically.
        prior_logins = list(dict.fromkeys(prior_logins))
        new_logins = [entry["github_login"] for entry in new_content]
        if attribution.kind == "user" and set(prior_logins) != set(new_logins):
            log_activity(
                organization_id=None,
                team_id=team.id,
                user=cast(User, request.user),
                was_impersonated=is_impersonated_session(request),
                item_id=report_id,
                scope="SignalReport",
                activity="suggested_reviewers_changed",
                detail=Detail(
                    name=report.title,
                    changes=[
                        Change(
                            type="SignalReport",
                            action="changed",
                            field="suggested_reviewers",
                            before=prior_logins,
                            after=new_logins,
                        )
                    ],
                ),
            )

            # A human added reviewers: ping the newly-added ones on their own Slack channel so
            # someone added after generation still hears about an actionable report, mirroring
            # the notification sent when it first went ready. Removals aren't notified.
            prior_login_set = set(prior_logins)
            added_logins = [login for login in new_logins if login not in prior_login_set]
            if added_logins:
                _schedule_reviewer_added_slack_notifications(
                    team_id=team.id,
                    report_id=str(report_id),
                    added_logins=added_logins,
                    actor_user_id=attribution.user_id,
                )

    return new_artefact, seen


@extend_schema_view(
    list=extend_schema(
        summary="List a report's artefacts",
        description=(
            "List every artefact on a report — the full work log: signal findings (the evidence "
            "behind the report), status judgments (safety / actionability / priority, repo "
            "selection, suggested reviewers — the newest row of each status type is canonical), "
            "and log entries (code references, commits, task runs, notes). "
            "`suggested_reviewers` content is enriched with Insights user info at read time."
        ),
        parameters=[_REPORT_ID_PARAMETER],
        responses={200: SignalReportArtefactSerializer(many=True)},
        operation_id="signals_report_artefacts_list",
    ),
    retrieve=extend_schema(
        summary="Get a single artefact",
        description="Get one artefact by id, content parsed (and reviewers enriched) the same way as the list.",
        parameters=[_REPORT_ID_PARAMETER],
        responses={200: SignalReportArtefactSerializer},
        operation_id="signals_report_artefacts_retrieve",
    ),
    # The bespoke reviewers PUT stays app-only: agents append suggested_reviewers via POST instead.
    update=extend_schema(exclude=True),
)
class SignalReportArtefactViewSet(
    TeamAndOrgViewSetMixin,
    mixins.ListModelMixin,
    mixins.RetrieveModelMixin,
    mixins.CreateModelMixin,
    mixins.UpdateModelMixin,
    mixins.DestroyModelMixin,
    viewsets.GenericViewSet,
):
    """Artefacts attached to a signal report.

    Two write surfaces, both gated by the `task:write` scope (already held by the agent tokens):

    - PUT edits a report's suggested reviewers: it appends a new `suggested_reviewers` status
      artefact (latest-wins, so the new row becomes current) with bespoke reviewer enrichment,
      merging commits/names forward from the current reviewers. Other types return 400.
    - POST / PATCH / DELETE manage artefacts of *any* type — no type is writer-restricted.
      Log entries accumulate; status types (judgments, repo selection, suggested reviewers)
      are latest-wins, so appending a new version supersedes the previous one as the report's
      canonical status. Content is validated against the type's schema. Team scoping is
      enforced by `safely_get_queryset`, so an artefact id from another team / a deleted
      report 404s.

    Writes are attributed: to the task named by the `X-Insights-Task-Id` header (set automatically
    for sandbox agents) when present, else to the requesting user.
    """

    serializer_class = SignalReportArtefactSerializer
    authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication]
    permission_classes = [IsAuthenticated, APIScopePermission]
    scope_object = "task"
    # select_related: the read serializer renders `created_by` inline.
    queryset = SignalReportArtefact.objects.select_related("created_by").order_by("-created_at")
    http_method_names = ["get", "post", "put", "patch", "delete", "head", "options"]

    def _validated_report_id(self) -> str:
        # `report_id` addresses a report's UUID primary key. Agents sometimes pass a signal id
        # (e.g. `sig_praise`) instead, which the ORM can't coerce to a UUID — left to reach the
        # query it surfaces as a 500. Reject it up front as a clean 404 the caller can recover from.
        report_id = self.parents_query_dict["report_id"]
        try:
            uuid.UUID(str(report_id))
        except (ValueError, TypeError):
            raise NotFound()
        return report_id

    def safely_get_queryset(self, queryset):
        # Mirror SignalReportViewSet: a deleted parent report is unreachable, so
        # its artefacts must be too (otherwise a known UUID would bypass deletion).
        return queryset.filter(
            report_id=self._validated_report_id(),
            team=self.team,
        ).exclude(report__status=SignalReport.Status.DELETED)

    def list(self, request, *args, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())
        # Surface legacy `SignalReportTask` associations as synthetic `task_run` artefacts so a
        # report's research / implementation runs appear in the log even before the backfill has
        # converted its gate rows. Merged into the materialized log (de-duplicated against the real
        # task_run artefacts) and re-sorted newest-first so each legacy row lands at its original
        # timestamp, then paginated as one list so `count` and ordering both account for them.
        real_artefacts = list(queryset)
        synthetic = SignalReport.synthetic_legacy_task_run_artefacts(
            report_id=self.parents_query_dict["report_id"],
            team_id=self.team.id,
            existing_artefacts=real_artefacts,
        )
        log = (
            sorted([*real_artefacts, *synthetic], key=lambda a: a.created_at, reverse=True)
            if synthetic
            else real_artefacts
        )
        page = self.paginate_queryset(log)
        artefacts = list(page if page is not None else log)
        logins_union = normalized_github_logins_from_suggested_reviewer_artefacts(artefacts)
        login_map = resolve_org_github_login_to_users(self.team.id, logins_union) if logins_union else {}
        serializer = SignalReportArtefactSerializer(
            artefacts,
            many=True,
            context={
                **self.get_serializer_context(),
                "signals_github_login_to_user_map": login_map,
            },
        )
        if page is not None:
            return self.get_paginated_response(serializer.data)
        return Response(serializer.data)

    def update(self, request, *args, **kwargs):
        artefact = cast(SignalReportArtefact, self.get_object())

        # Generic endpoint, single-type allow-list: any other artefact type is
        # part of the agentic pipeline contract and must not be hand-edited.
        if artefact.type != SignalReportArtefact.ArtefactType.SUGGESTED_REVIEWERS:
            return Response(
                {
                    "error": (
                        "Only suggested_reviewers artefacts may be modified via this endpoint. "
                        f"This artefact has type '{artefact.type}'."
                    )
                },
                status=status.HTTP_400_BAD_REQUEST,
            )

        write_serializer = SignalReportArtefactWriteSerializer(data=request.data)
        write_serializer.is_valid(raise_exception=True)
        entries = write_serializer.validated_data["content"]

        try:
            new_artefact, seen = append_suggested_reviewers(
                team=self.team,
                report_id=str(artefact.report_id),
                entries=entries,
                request=request,
            )
        except ReviewerWriteError as e:
            return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)

        # Return the read-shape (enriched) so the client sees the canonical result.
        login_map = resolve_org_github_login_to_users(self.team.id, list(seen)) if seen else {}
        read_serializer = SignalReportArtefactSerializer(
            new_artefact,
            context={
                **self.get_serializer_context(),
                "signals_github_login_to_user_map": login_map,
            },
        )
        return Response(read_serializer.data)

    @staticmethod
    def _write_response_data(artefact: SignalReportArtefact) -> dict:
        """Build the create/update response payload, parsing stored JSON content for the echo."""
        try:
            parsed_content: dict | list = json.loads(artefact.content)
        except (json.JSONDecodeError, ValueError):
            parsed_content = {}
        return SignalReportArtefactWriteResponseSerializer(
            {
                "id": artefact.id,
                "report_id": artefact.report_id,
                "type": artefact.type,
                "content": parsed_content,
                "created_at": artefact.created_at,
                "updated_at": artefact.updated_at,
                "task_id": artefact.task_id,
            }
        ).data

    @validated_request(
        request_serializer=SignalReportArtefactLogCreateSerializer,
        responses={
            201: OpenApiResponse(response=SignalReportArtefactWriteResponseSerializer, description="Artefact created."),
            400: OpenApiResponse(
                description="Unknown artefact type, content not matching the type's schema, "
                "or an invalid X-Insights-Task-Id header."
            ),
            404: OpenApiResponse(description="Report not found for this project."),
        },
        parameters=[
            _REPORT_ID_PARAMETER,
            OpenApiParameter(
                name=TASK_ID_HEADER,
                type=OpenApiTypes.UUID,
                location=OpenApiParameter.HEADER,
                required=False,
                description=(
                    "Task to attribute the artefact to (must belong to this project). Set automatically "
                    "for sandbox agents; when absent the artefact is attributed to the requesting user."
                ),
            ),
        ],
        summary="Append an artefact to a report",
        description=(
            "Append an artefact to a report (see artefact_type for the writable types). Everything "
            "is append-only: log entries (code reference, commit, task run, note) accumulate, while "
            "status types (safety / actionability / priority judgments, repo selection, suggested "
            "reviewers) are latest-wins — appending a new version supersedes the previous one as the "
            "report's canonical status. Content is validated against the type's schema."
        ),
        operation_id="signals_report_artefacts_create",
    )
    def create(self, request: ValidatedRequest, *args, **kwargs) -> Response:
        report_id = self._validated_report_id()
        # A deleted / foreign report is unreachable — don't let a known report_id attach
        # artefacts to it. Mirrors the team-scoped filter in `safely_get_queryset`.
        report_exists = (
            SignalReport.objects.filter(id=report_id, team=self.team)
            .exclude(status=SignalReport.Status.DELETED)
            .exists()
        )
        if not report_exists:
            raise NotFound()
        artefact_type = request.validated_data["artefact_type"]
        writable_types = sorted(set(SignalReportArtefact.ArtefactType.values) - NON_WRITABLE_ARTEFACT_TYPES)
        if artefact_type not in SignalReportArtefact.ArtefactType.values:
            return Response(
                {"error": f"Unknown artefact type '{artefact_type}'. Valid types: {', '.join(writable_types)}."},
                status=status.HTTP_400_BAD_REQUEST,
            )
        if artefact_type in NON_WRITABLE_ARTEFACT_TYPES:
            # Legacy permissive types (e.g. video_segment) have no real content validation, so the
            # write API refuses them — they stay readable for existing rows but can't be created.
            return Response(
                {"error": f"Artefact type '{artefact_type}' is read-only and cannot be created through the API."},
                status=status.HTTP_400_BAD_REQUEST,
            )
        content = request.validated_data["content"]
        attribution = resolve_request_attribution(request, self.team.id)
        if artefact_type == SignalReportArtefact.ArtefactType.TASK_RUN:
            # task_run artefacts are the task↔report association itself, so they get
            # associate-me ergonomics: content.task_id defaults to the calling agent's task
            # (the header), product/type default to a generic agent-run label, the named task
            # must belong to this project, attribution is always the recorded task, and
            # re-associating an already-linked task is idempotent (returns the existing entry).
            if not isinstance(content, dict):
                return Response(
                    {"error": "task_run content must be a JSON object."},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            task_id = content.get("task_id") or resolve_task_id_from_header(request, self.team.id)
            if not task_id:
                return Response(
                    {"error": "Provide content.task_id, or call with an X-Insights-Task-Id header."},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            if not tasks_facade.task_exists(task_id, self.team.id):
                return Response(
                    {"error": "Unknown task for this project."},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            content = {**content, "task_id": str(task_id)}
            content.setdefault("product", "tasks")
            content.setdefault("type", "agent_run")
            existing = (
                SignalReportArtefact.objects.filter(
                    team=self.team,
                    report_id=report_id,
                    type=SignalReportArtefact.ArtefactType.TASK_RUN,
                    task_id=task_id,
                )
                .order_by("created_at")
                .first()
            )
            if existing is not None:
                return Response(self._write_response_data(existing), status=status.HTTP_200_OK)
            attribution = ArtefactAttribution.from_task(str(task_id))
        # The write boundary: parse the raw payload into the type's content model once; the
        # typed model is what flows into the append helpers.
        try:
            parsed_content = parse_artefact_content(artefact_type, content)
        except ArtefactContentValidationError as e:
            return Response(
                {"error": f"content does not match the '{artefact_type}' schema: {e}"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        artefact = SignalReportArtefact.append(
            team_id=self.team.id,
            report_id=report_id,
            content=parsed_content,
            attribution=attribution,
        )
        return Response(self._write_response_data(artefact), status=status.HTTP_201_CREATED)

    @validated_request(
        request_serializer=SignalReportArtefactLogUpdateSerializer,
        responses={
            200: OpenApiResponse(response=SignalReportArtefactWriteResponseSerializer, description="Artefact updated."),
            400: OpenApiResponse(description="Content does not match the artefact type's schema."),
            404: OpenApiResponse(description="Artefact not found for this report / project."),
        },
        summary="Replace an artefact's content",
        description=(
            "Replace the content of an existing artefact, addressed by id. The new content is "
            "validated against the artefact's type schema. Editing the latest row of a status type "
            "changes the report's canonical status (latest-wins); to re-assess while keeping history, "
            "append a new artefact instead. Attribution is creation-time only — edits don't reassign it."
        ),
        parameters=[_REPORT_ID_PARAMETER],
        operation_id="signals_report_artefacts_partial_update",
    )
    def partial_update(self, request: ValidatedRequest, *args, **kwargs) -> Response:
        artefact = cast(SignalReportArtefact, self.get_object())
        if artefact.type in NON_WRITABLE_ARTEFACT_TYPES:
            # Legacy read-only types (e.g. video_segment) can't be created via the API, so they
            # can't be edited through it either.
            return Response(
                {"error": f"Artefact type '{artefact.type}' is read-only and cannot be edited through the API."},
                status=status.HTTP_400_BAD_REQUEST,
            )
        try:
            artefact.update_content(request.validated_data["content"])
        except ArtefactContentValidationError as e:
            return Response(
                {"error": f"content does not match the '{artefact.type}' schema: {e}"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        return Response(self._write_response_data(artefact))

    @extend_schema(
        responses={
            204: OpenApiResponse(description="Artefact deleted."),
            404: OpenApiResponse(description="Artefact not found for this report / project."),
        },
        summary="Delete an artefact",
        description=(
            "Delete an artefact, addressed by id. Deleting the latest row of a status type reverts "
            "the report's canonical status to the previous version (latest-wins over what remains)."
        ),
        parameters=[_REPORT_ID_PARAMETER],
        operation_id="signals_report_artefacts_destroy",
    )
    def destroy(self, request, *args, **kwargs) -> Response:
        artefact = cast(SignalReportArtefact, self.get_object())
        artefact.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

    @extend_schema(
        responses={
            200: OpenApiResponse(
                response=CommitDiffResponseSerializer,
                description="The branch's unified diff against the repository default branch.",
            ),
            400: OpenApiResponse(description="Artefact is not a commit, or is missing repository/branch."),
            404: OpenApiResponse(description="Artefact not found, or no GitHub integration can access the repository."),
            502: OpenApiResponse(description="GitHub could not produce the diff (branch not found, fetch failed)."),
        },
        summary="Fetch the diff for a commit artefact",
        description=(
            "Fetch the unified diff of a `commit` artefact's branch against the repository default "
            "branch via the team's GitHub integration — using the branch's current tip so the diff "
            "reflects the latest state of the work, not just the single recorded commit."
        ),
        parameters=[_REPORT_ID_PARAMETER],
        operation_id="signals_report_artefacts_diff",
    )
    @action(detail=True, methods=["get"], url_path="diff", required_scopes=["task:read"])
    def diff(self, request: Request, *args, **kwargs) -> Response:
        artefact = cast(SignalReportArtefact, self.get_object())
        if artefact.type != SignalReportArtefact.ArtefactType.COMMIT:
            return Response(
                {"error": f"Diffs are only available for commit artefacts, not '{artefact.type}'."},
                status=status.HTTP_400_BAD_REQUEST,
            )
        try:
            content = json.loads(artefact.content)
        except (json.JSONDecodeError, ValueError):
            content = {}
        if not isinstance(content, dict):
            # Log artefacts store arbitrary JSON; a non-object payload has no repository/commit.
            content = {}
        repository = content.get("repository")
        branch = content.get("branch")
        if not repository or not branch:
            return Response(
                {"error": "Artefact is missing a repository or branch."},
                status=status.HTTP_400_BAD_REQUEST,
            )

        # The diff is deliberately bounded to repos the team's GitHub installation can access
        # (`first_for_team_repository` returns None otherwise) rather than to a single per-report
        # repository: a report's work legitimately spans multiple repos (cross-repo fixes, stacked
        # PRs). That connection boundary is the intended scope — any `task:write` holder can already
        # run agents against those same repos — and the repo/ref values are validated in `get_diff`.
        try:
            github = GitHubIntegration.first_for_team_repository(self.team.id, repository)
        except GitHubRateLimitError as e:
            return github_rate_limited_response(e)
        if github is None:
            return Response(
                {"error": f"No GitHub integration can access '{repository}'."},
                status=status.HTTP_404_NOT_FOUND,
            )
        try:
            # Diff the commit's branch against the repo default branch, using each branch's current
            # tip (no SHA pinning) so the diff stays useful as the branch keeps moving after the
            # commit was recorded — e.g. after PR babysitting or customer tweaks.
            base_branch = github.get_default_branch(repository)
            result = github.get_diff(repository, target_branch=str(branch), base_branch=base_branch)
        except GitHubRateLimitError as e:
            return github_rate_limited_response(e)
        except Exception:  # noqa: BLE001 — never let an upstream GitHub failure 500 this endpoint
            logger.warning(
                "signals branch diff fetch errored",
                repository=repository,
                branch=branch,
            )
            return Response(
                {"error": "GitHub could not produce the diff for this branch."},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        if not result.get("success"):
            # Surface a clean message rather than the raw GitHub error body. A 404 from the
            # compare API means the branch (or repo) isn't on the remote — most often a branch
            # that was merged-and-deleted or force-rewritten away.
            if result.get("status_code") == 404:
                return Response(
                    {
                        "error": f"Branch '{branch}' or repository '{repository}' was not found on GitHub — "
                        "the branch may have been deleted or merged away."
                    },
                    status=status.HTTP_404_NOT_FOUND,
                )
            logger.warning(
                "signals branch diff fetch failed",
                repository=repository,
                branch=branch,
                status_code=result.get("status_code"),
            )
            return Response(
                {"error": "GitHub could not produce the diff for this branch."},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        return Response(
            {
                "diff": result["diff"],
                "truncated": result.get("truncated", False),
            }
        )


class SignalUserAutonomyConfigView(APIView):
    """Per-user signal autonomy config (singleton keyed by user).

    GET    /v1/users/<id>/signal_autonomy/ → current config (or 404)
    POST   /v1/users/<id>/signal_autonomy/ → create or update
    DELETE /v1/users/<id>/signal_autonomy/ → remove (opt out)
    """

    serializer_class = SignalUserAutonomyConfigSerializer
    authentication_classes = [SessionAuthentication, PersonalAPIKeyAuthentication, OAuthAccessTokenAuthentication]
    permission_classes = [IsAuthenticated, APIScopePermission]
    scope_object = "user"
    required_scopes = ["user:write"]

    def _resolve_user(self, request, user_id):
        if str(user_id) == "@me":
            return request.user
        if not request.user.is_staff:
            raise exceptions.PermissionDenied("Only staff can access other users' autonomy config.")

        try:
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            raise exceptions.NotFound()

    @extend_schema(responses={200: SignalUserAutonomyConfigSerializer})
    def get(self, request, user_id, **kwargs):
        user = self._resolve_user(request, user_id)
        config = SignalUserAutonomyConfig.objects.filter(user=user).first()
        if config is None:
            return Response(status=status.HTTP_404_NOT_FOUND)
        return Response(SignalUserAutonomyConfigSerializer(config).data)

    @extend_schema(responses={200: SignalUserAutonomyConfigSerializer})
    def post(self, request, user_id, **kwargs):
        user = self._resolve_user(request, user_id)
        serializer = SignalUserAutonomyConfigCreateSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        validated = serializer.validated_data
        # Partial update: only touch fields the client explicitly sent. This lets the
        # auto-start setting and the Slack notification settings be edited
        # independently from separate UI surfaces without one wiping the other.
        defaults: dict = {}
        if "autostart_priority" in serializer.initial_data:
            defaults["autostart_priority"] = validated.get("autostart_priority")
        if "slack_notification_min_priority" in serializer.initial_data:
            defaults["slack_notification_min_priority"] = validated.get("slack_notification_min_priority")
        if "slack_notification_channel" in serializer.initial_data:
            defaults["slack_notification_channel"] = validated.get("slack_notification_channel") or None
        if "slack_notification_integration_id" in serializer.initial_data:
            integration_id = validated.get("slack_notification_integration_id")
            integration = None
            if integration_id is not None:
                current_team_id = request.user.current_team_id
                if current_team_id is None or current_team_id not in UserPermissions(user).team_ids_visible_for_user:
                    raise serializers.ValidationError(
                        {"slack_notification_integration_id": "Unknown Slack integration for this user."}
                    )
                candidate = Integration.objects.filter(
                    pk=integration_id,
                    kind="slack",
                    team_id=current_team_id,
                ).first()
                if candidate is None:
                    raise serializers.ValidationError(
                        {"slack_notification_integration_id": "Unknown Slack integration for this user."}
                    )
                integration = candidate
            defaults["slack_notification_integration"] = integration
        config, _created = SignalUserAutonomyConfig.objects.update_or_create(
            user=user,
            defaults=defaults,
        )
        return Response(SignalUserAutonomyConfigSerializer(config).data)

    @extend_schema(responses={204: None})
    def delete(self, request, user_id, **kwargs):
        user = self._resolve_user(request, user_id)
        SignalUserAutonomyConfig.objects.filter(user=user).delete()
        return Response(status=status.HTTP_204_NO_CONTENT)


class PauseUntilRequestSerializer(serializers.Serializer):
    timestamp = serializers.DateTimeField(help_text="Pause the grouping pipeline until this timestamp (ISO 8601).")


class PauseResponseSerializer(serializers.Serializer):
    status = serializers.CharField(help_text="Always 'paused'.")
    paused_until = serializers.DateTimeField(help_text="The timestamp the pipeline is paused until.")


class UnpauseResponseSerializer(serializers.Serializer):
    status = serializers.CharField(help_text="Always 'unpaused'.")
    was_paused = serializers.BooleanField(help_text="Whether the workflow was actually paused at the time of the call.")


class PauseStateResponseSerializer(serializers.Serializer):
    paused_until = serializers.DateTimeField(
        allow_null=True, help_text="The timestamp the pipeline is paused until, or null if not paused/not running."
    )


class SignalProcessingViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet):
    """View and control signal processing pipeline state for a team."""

    # Same scope family as other signals team APIs (SignalSourceConfigViewSet, etc.)
    scope_object = "task"
    scope_object_write_actions = [
        "create",
        "update",
        "partial_update",
        "patch",
        "destroy",
        "pause",
        "unpause",
    ]

    @extend_schema(request=None, responses={200: PauseStateResponseSerializer})
    def list(self, request: Request, *args, **kwargs) -> Response:
        """Return current processing state including pause status."""
        state = async_to_sync(TeamSignalGroupingV2Workflow.paused_state)(self.team.id)
        return Response({"paused_until": state.isoformat() if state else None})

    @extend_schema(request=PauseUntilRequestSerializer, responses={200: PauseResponseSerializer})
    @action(methods=["PUT", "DELETE"], detail=False, url_path="pause")
    def pause(self, request: Request, *args, **kwargs) -> Response:
        if request.method == "DELETE":
            was_paused = async_to_sync(TeamSignalGroupingV2Workflow.unpause)(self.team.id)
            return Response({"status": "unpaused", "was_paused": was_paused})

        serializer = PauseUntilRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        timestamp = serializer.validated_data["timestamp"]
        async_to_sync(TeamSignalGroupingV2Workflow.pause_until)(self.team.id, timestamp)
        return Response({"status": "paused", "paused_until": timestamp.isoformat()})