ref(seer): Type 10 issue/event/agent/PR seer RPC responses

Replaces the `dict[str, Any]` return annotations on the following seer RPC
methods with explicit Pydantic models so the seer-side SDK consumer sees
declared field shapes:

- `get_issue_details` → `IssueDetailsResponse | None`
- `get_event_details` → `EventDetailsResponse | None`
- `get_issue_and_event_details_v2` → `IssueAndEventDetailsResponse | None`
  (single model whose issue-side fields use `exclude_unset` so they're
  absent from the wire when `include_issue=False` or the group lookup
  failed — matches the pre-typed conditional spread)
- `get_transactions_for_project` → `TransactionsForProjectResponse`
- `get_trace_for_transaction` → `TraceData | EmptyResponse`
- `get_profiles_for_trace` → `TraceProfiles | EmptyResponse`
- `get_issues_for_transaction` → `TransactionIssues | EmptyResponse`
  (the four project-scoped RPC wrappers above already built the response
  by calling `.dict()` on existing Pydantic models — just return them
  directly and use `EmptyResponse` for the `{}` not-found case)
- `bulk_get_project_preferences` → `BulkProjectPreferencesResponse`
  (bare `{project_id_str: pref_dict}` map via `__root__` passthrough)
- `record_pr_attribution` → `PrAttributionResponse`
- `update_pr_metrics` → `UpdatePrMetricsSuccessResponse | UpdatePrMetricsErrorResponse`
  (discriminated by `success: Literal[True|False]`)

Wire-identical across all paths. The response models that need to be
read like dicts by seer or existing test sites carry a small `_DictProxyMixin`
(`__getitem__`/`__contains__`/`.get`) so the typed return doesn't force
a rewrite of every consumer in the same change.

Author: azulus

src/sentry/pr_metrics/judge.py

@@ -39,6 +39,10 @@
 from sentry.pr_metrics.utils import iso_or_none, resolved_group_ids
 from sentry.seer.code_review.models import SeerCodeReviewRepoDefinition
 from sentry.seer.code_review.utils import build_repo_definition
+from sentry.seer.sentry_data_models import (
+    UpdatePrMetricsErrorResponse,
+    UpdatePrMetricsSuccessResponse,
+)
 from sentry.seer.signed_seer_api import SeerViewerContext, make_signed_seer_api_request
 from sentry.utils import metrics
 
@@ -245,7 +249,7 @@ def update_pr_metrics(
     repository_id: int,
     verdict: str | None = None,
     attributions: Sequence[Mapping[str, Any]] | None = None,
-) -> dict[str, Any]:
+) -> UpdatePrMetricsSuccessResponse | UpdatePrMetricsErrorResponse:
     """Persist Seer's judge result for a PR and emit the enriched metrics row.
 
     Inbound Seer RPC (Seer → Sentry), invoked once Seer has judged a forwarded
@@ -278,14 +282,14 @@ def update_pr_metrics(
     if verdict is None or verdict not in RESULT_VERDICTS:
         logger.warning("pr_metrics.update.invalid_verdict", extra={**log_extra, "verdict": verdict})
         metrics.incr("pr_metrics.update.skipped", tags={"reason": "invalid_verdict"})
-        return {"success": False, "error": "invalid_verdict"}
+        return UpdatePrMetricsErrorResponse(error="invalid_verdict")
 
     try:
         parsed_attributions = _parse_attributions(attributions or ())
     except (KeyError, TypeError, ValueError):
         logger.warning("pr_metrics.update.invalid_attribution", extra=log_extra)
         metrics.incr("pr_metrics.update.skipped", tags={"reason": "invalid_attribution"})
-        return {"success": False, "error": "invalid_attribution"}
+        return UpdatePrMetricsErrorResponse(error="invalid_attribution")
 
     # Scope the lookup to the reported org+repo: the id alone is attacker-influenced
     # (it round-trips through Seer), so trusting it unscoped would be an IDOR.
@@ -298,15 +302,15 @@ def update_pr_metrics(
     except PullRequest.DoesNotExist:
         logger.warning("pr_metrics.update.pull_request_not_found", extra=log_extra)
         metrics.incr("pr_metrics.update.skipped", tags={"reason": "pr_not_found"})
-        return {"success": False, "error": "pull_request_not_found"}
+        return UpdatePrMetricsErrorResponse(error="pull_request_not_found")
 
     # Emit needs a terminal PR (closed_at + head_commit_sha). Validate it before
     # writing so a non-terminal PR is rejected up front rather than committing the
     # verdict and then failing in emit — i.e. no committed-but-errored state.
     if pull_request.closed_at is None or pull_request.head_commit_sha is None:
         logger.warning("pr_metrics.update.not_terminal", extra=log_extra)
         metrics.incr("pr_metrics.update.skipped", tags={"reason": "not_terminal"})
-        return {"success": False, "error": "pull_request_not_terminal"}
+        return UpdatePrMetricsErrorResponse(error="pull_request_not_terminal")
 
     # Only the verdict is written here; the webhook keeps the activity counters
     # current, so this partial update must not clobber them.
@@ -328,7 +332,7 @@ def update_pr_metrics(
                 "pr_metrics.update.already_settled", extra={**log_extra, "verdict": verdict}
             )
             metrics.incr("pr_metrics.update.skipped", tags={"reason": "already_settled"})
-            return {"success": True}
+            return UpdatePrMetricsSuccessResponse()
         for signal_type, source, signal_details in parsed_attributions:
             record_attribution_signal(
                 pull_request=pull_request,
@@ -341,4 +345,4 @@ def update_pr_metrics(
 
     metrics.incr("pr_metrics.update.recorded", tags={"verdict": verdict})
     logger.info("pr_metrics.update.recorded", extra={**log_extra, "verdict": verdict})
-    return {"success": True}
+    return UpdatePrMetricsSuccessResponse()

src/sentry/seer/agent/index_data.py

@@ -20,13 +20,15 @@
     normalize_description,
 )
 from sentry.seer.sentry_data_models import (
+    EmptyResponse,
     IssueDetails,
     ProfileData,
     Span,
     TraceData,
     TraceProfiles,
     Transaction,
     TransactionIssues,
+    TransactionsForProjectResponse,
 )
 from sentry.services.eventstore import backend as eventstore
 from sentry.services.eventstore.models import Event, GroupEvent
@@ -540,22 +542,25 @@ def get_issues_for_transaction(transaction_name: str, project_id: int) -> Transa
 # RPC wrappers
 
 
-def rpc_get_transactions_for_project(project_id: int) -> dict[str, Any]:
+def rpc_get_transactions_for_project(project_id: int) -> TransactionsForProjectResponse:
     transactions = get_transactions_for_project(project_id)
-    transaction_dicts = [transaction.dict() for transaction in transactions]
-    return {"transactions": transaction_dicts}
+    return TransactionsForProjectResponse(transactions=list(transactions))
 
 
-def rpc_get_trace_for_transaction(transaction_name: str, project_id: int) -> dict[str, Any]:
+def rpc_get_trace_for_transaction(
+    transaction_name: str, project_id: int
+) -> TraceData | EmptyResponse:
     trace = get_trace_for_transaction(transaction_name, project_id)
-    return trace.dict() if trace else {}
+    return trace if trace is not None else EmptyResponse()
 
 
-def rpc_get_profiles_for_trace(trace_id: str, project_id: int) -> dict[str, Any]:
+def rpc_get_profiles_for_trace(trace_id: str, project_id: int) -> TraceProfiles | EmptyResponse:
     profiles = get_profiles_for_trace(trace_id, project_id)
-    return profiles.dict() if profiles else {}
+    return profiles if profiles is not None else EmptyResponse()
 
 
-def rpc_get_issues_for_transaction(transaction_name: str, project_id: int) -> dict[str, Any]:
+def rpc_get_issues_for_transaction(
+    transaction_name: str, project_id: int
+) -> TransactionIssues | EmptyResponse:
     issues = get_issues_for_transaction(transaction_name, project_id)
-    return issues.dict() if issues else {}
+    return issues if issues is not None else EmptyResponse()

src/sentry/seer/agent/tools.py

@@ -61,9 +61,12 @@
 from sentry.seer.sentry_data_models import (
     EAPTrace,
     EmptyResponse,
+    EventDetailsResponse,
     ExecuteQueryErrorResponse,
     ExecuteQuerySuccessResponse,
     GetDsnResponse,
+    IssueAndEventDetailsResponse,
+    IssueDetailsResponse,
     RepositoryDefinitionResponse,
     TraceItemAttributesResponse,
     TraceItemEventsResponse,
@@ -1245,100 +1248,100 @@ def get_issue_and_event_response(
     organization: Organization,
     start: datetime | None = None,
     end: datetime | None = None,
-) -> dict[str, Any]:
+) -> IssueAndEventDetailsResponse:
     serialized_event = dict(serialize(event, user=None, serializer=EventSerializer()))
     serialized_event.update(_get_event_troubleshooting_context(event))
 
-    result = {
+    event_fields: dict[str, Any] = {
         "event": serialized_event,
         "event_id": event.event_id,
         "event_trace_id": event.trace_id,
         "project_id": event.project_id,
         "project_slug": event.project.slug,
     }
 
-    if group is not None:
-        # Get the issue metadata, tags overview, and event count timeseries.
-        serialized_group = dict(serialize(group, user=None, serializer=GroupSerializer()))
-        # Add issueTypeDescription as it provides better context for LLMs. Note the initial type should be BaseGroupSerializerResponse.
-        serialized_group["issueTypeDescription"] = group.issue_type.description
+    if group is None:
+        return IssueAndEventDetailsResponse(**event_fields)
 
-        logger.info(
-            "get_issue_and_event_details_v2: Querying for tags overview",
+    # Get the issue metadata, tags overview, and event count timeseries.
+    serialized_group = dict(serialize(group, user=None, serializer=GroupSerializer()))
+    # Add issueTypeDescription as it provides better context for LLMs. Note the initial type should be BaseGroupSerializerResponse.
+    serialized_group["issueTypeDescription"] = group.issue_type.description
+
+    logger.info(
+        "get_issue_and_event_details_v2: Querying for tags overview",
+        extra={
+            "organization_id": organization.id,
+            "issue_id": group.id,
+            "timedelta": (end - start) if start and end else None,
+            "start": start,
+            "end": end,
+        },
+    )
+
+    try:
+        tags_overview = get_all_tags_overview(group, start, end)
+    except Exception:
+        logger.exception(
+            "Failed to get tags overview for issue",
             extra={
                 "organization_id": organization.id,
                 "issue_id": group.id,
-                "timedelta": (end - start) if start and end else None,
                 "start": start,
                 "end": end,
             },
         )
+        tags_overview = None
 
-        try:
-            tags_overview = get_all_tags_overview(group, start, end)
-        except Exception:
-            logger.exception(
-                "Failed to get tags overview for issue",
-                extra={
-                    "organization_id": organization.id,
-                    "issue_id": group.id,
-                    "start": start,
-                    "end": end,
-                },
-            )
-            tags_overview = None
-
-        try:
-            ts_result = _get_issue_event_timeseries(
-                group=group,
-                organization=organization,
-                start=start,
-                end=end,
-            )
-        except Exception:
-            logger.exception(
-                "Failed to get issue event timeseries",
-                extra={
-                    "organization_id": organization.id,
-                    "issue_id": group.id,
-                    "start": start,
-                    "end": end,
-                },
-            )
-            ts_result = None
+    try:
+        ts_result = _get_issue_event_timeseries(
+            group=group,
+            organization=organization,
+            start=start,
+            end=end,
+        )
+    except Exception:
+        logger.exception(
+            "Failed to get issue event timeseries",
+            extra={
+                "organization_id": organization.id,
+                "issue_id": group.id,
+                "start": start,
+                "end": end,
+            },
+        )
+        ts_result = None
 
-        if ts_result:
-            timeseries, timeseries_stats_period, timeseries_interval = ts_result
-        else:
-            timeseries, timeseries_stats_period, timeseries_interval = None, None, None
+    if ts_result:
+        timeseries, timeseries_stats_period, timeseries_interval = ts_result
+    else:
+        timeseries, timeseries_stats_period, timeseries_interval = None, None, None
 
-        # Fetch user activity (comments, status changes, etc.)
-        try:
-            activities = Activity.objects.filter(
-                group=group,
-                type__in=_SEER_EXPLORER_ACTIVITY_TYPES,
-            ).order_by("-datetime")[:50]
-            serialized_activities = serialize(
-                list(activities), user=None, serializer=ActivitySerializer()
-            )
-        except Exception:
-            logger.exception(
-                "Failed to get user activity for issue",
-                extra={"organization_id": organization.id, "issue_id": group.id},
-            )
-            serialized_activities = []
-
-        result = {
-            **result,
-            "issue": serialized_group,
-            "event_timeseries": timeseries,
-            "timeseries_stats_period": timeseries_stats_period,
-            "timeseries_interval": timeseries_interval,
-            "tags_overview": tags_overview,
-            "user_activity": serialized_activities,
-        }
+    # Fetch user activity (comments, status changes, etc.)
+    try:
+        activities = Activity.objects.filter(
+            group=group,
+            type__in=_SEER_EXPLORER_ACTIVITY_TYPES,
+        ).order_by("-datetime")[:50]
+        serialized_activities = serialize(
+            list(activities), user=None, serializer=ActivitySerializer()
+        )
+    except Exception:
+        logger.exception(
+            "Failed to get user activity for issue",
+            extra={"organization_id": organization.id, "issue_id": group.id},
+        )
+        serialized_activities = []
 
-    return result
+    return IssueAndEventDetailsResponse(
+        **event_fields,
+        issue=serialized_group,
+        event_timeseries=timeseries,
+        timeseries_stats_period=timeseries_stats_period,
+        timeseries_interval=timeseries_interval,
+        tags_overview=tags_overview,
+        user_activity=serialized_activities,
+    )
 
 
 def get_issue_details(
@@ -1348,7 +1351,7 @@ def get_issue_details(
     start: str | None = None,
     end: str | None = None,
     project_slug: str | None = None,
-) -> dict[str, Any] | None:
+) -> IssueDetailsResponse | None:
     """
     Get issue-level details for an issue, optionally scoped by time range.
 
@@ -1433,16 +1436,16 @@ def get_issue_details(
         )
         serialized_activities = []
 
-    return {
-        "issue": serialized_group,
-        "event_timeseries": timeseries,
-        "timeseries_stats_period": timeseries_stats_period,
-        "timeseries_interval": timeseries_interval,
-        "tags_overview": tags_overview,
-        "user_activity": serialized_activities,
-        "project_id": group.project_id,
-        "project_slug": group.project.slug,
-    }
+    return IssueDetailsResponse(
+        issue=serialized_group,
+        event_timeseries=timeseries,
+        timeseries_stats_period=timeseries_stats_period,
+        timeseries_interval=timeseries_interval,
+        tags_overview=tags_overview,
+        user_activity=serialized_activities,
+        project_id=group.project_id,
+        project_slug=group.project.slug,
+    )
 
 
 def get_event_details(
@@ -1453,7 +1456,7 @@ def get_event_details(
     start: str | None = None,
     end: str | None = None,
     project_slug: str | None = None,
-) -> dict[str, Any] | None:
+) -> EventDetailsResponse | None:
     """
     Get event details by event ID, or get the recommended event for an issue, optionally scoped by time range.
     Exactly one of event_id or issue_id must be provided.
@@ -1554,13 +1557,13 @@ def get_event_details(
     serialized_event = dict(serialize(event, user=None, serializer=EventSerializer()))
     serialized_event.update(_get_event_troubleshooting_context(event))
 
-    return {
-        "event": serialized_event,
-        "event_id": event.event_id,
-        "event_trace_id": event.trace_id,
-        "project_id": event.project_id,
-        "project_slug": event.project.slug,
-    }
+    return EventDetailsResponse(
+        event=serialized_event,
+        event_id=event.event_id,
+        event_trace_id=event.trace_id,
+        project_id=event.project_id,
+        project_slug=event.project.slug,
+    )
 
 
 def get_issue_and_event_details_v2(
@@ -1572,7 +1575,7 @@ def get_issue_and_event_details_v2(
     event_id: str | None = None,
     project_slug: str | None = None,
     include_issue: bool = True,
-) -> dict[str, Any] | None:
+) -> IssueAndEventDetailsResponse | None:
     if bool(issue_id) == bool(event_id):
         raise BadRequest("Either issue_id or event_id must be provided, but not both.")