feat(seer): Add dual-read helpers for Seer project preferences from Sentry DB (#111591)

relates to AIML-2611, AIML-2610

Add `read_preference_from_sentry_db()` and
`bulk_read_preferences_from_sentry_db()` helpers to read Seer project
preferences directly from Sentry DB instead of Seer API. This is the
foundation for cutting over reads behind the
`organizations:seer-project-settings-read-from-sentry` feature flag.

We don't actually use the helpers in this PR. WIP followup PR:
#111594

### Extra

We return None if a project is unconfigured. Matching Seer's behavior
([single](https://github.com/getsentry/seer/blob/3cec03cac3abd9f1f26ac42296d6c870669c66f0/src/seer/automation/preferences.py#L54),
[bulk](https://github.com/getsentry/seer/blob/3cec03cac3abd9f1f26ac42296d6c870669c66f0/src/seer/automation/preferences.py#L98)),
"unconfigured" means we don't have any SeerProjectRepository rows _and_
none of the project options are set.

We use `project.get_option` and `ProjectOptions.objects.isset` for
single reads, and `ProjectOption.objects.get_value_bulk_id` for bulk
read. Why:
1. `project.get_option` and `ProjectOptions.objects.isset` use cache.
For a single project we want to check the cache first.
2. `ProjectOption.objects.get_value_bulk_id` is one DB query. For our
bulk use cases (`OrganizationAutofixAutomationSettingsEndpoint`
get/post, `configure_seer_for_existing_org`) where we may be querying
many projects for a single org, we do max 5 option queries, one for each
project option. It also returns None for missing options so we can
easily tell which projects have configured options and which don't.

---------

Co-authored-by: Claude Sonnet 4 <noreply@anthropic.com>
Co-authored-by: Cursor Agent <cursoragent@cursor.com>

Author: 3 people

src/sentry/features/temporary.py

@@ -382,6 +382,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:data-browsing-widget-unfurl", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable dual-write of Seer project preferences to Sentry DB and Seer API
     manager.add("organizations:seer-project-settings-dual-write", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enable reading of Seer project preferences from Sentry DB instead of Seer API
+    manager.add("organizations:seer-project-settings-read-from-sentry", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable public RPC endpoint for local seer development
     manager.add("organizations:seer-public-rpc", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Organizations on the old usage-based (v0) Seer plan

src/sentry/projectoptions/defaults.py

@@ -196,6 +196,14 @@
 register(key="sentry:seer_automation_handoff_integration_id", default=None)
 register(key="sentry:seer_automation_handoff_auto_create_pr", default=False)
 
+SEER_PROJECT_PREFERENCE_OPTION_KEYS = [
+    "sentry:seer_automated_run_stopping_point",
+    "sentry:seer_automation_handoff_point",
+    "sentry:seer_automation_handoff_target",
+    "sentry:seer_automation_handoff_integration_id",
+    "sentry:seer_automation_handoff_auto_create_pr",
+]
+
 # Boolean to enable/disable preprod size analysis for this project.
 register(key="sentry:preprod_size_enabled_by_customer", default=True)
 

src/sentry/seer/autofix/utils.py

@@ -1,19 +1,21 @@
 import logging
-from collections.abc import Iterable
+from collections import defaultdict
+from collections.abc import Callable, Iterable, Mapping
 from datetime import UTC, datetime
 from enum import StrEnum
 from typing import Any, NotRequired, TypedDict
 
 import orjson
 import pydantic
+import sentry_sdk
 from django.conf import settings
 from django.db import router, transaction
 from pydantic import BaseModel
 from rest_framework import serializers
 from urllib3 import BaseHTTPResponse, HTTPConnectionPool
 from urllib3.util.retry import Retry
 
-from sentry import features, options, ratelimits
+from sentry import features, options, projectoptions, ratelimits
 from sentry.constants import (
     AUTO_OPEN_PRS_DEFAULT,
     SEER_AUTOMATED_RUN_STOPPING_POINT_DEFAULT,
@@ -25,16 +27,20 @@
     get_sorted_code_mapping_configs,
 )
 from sentry.models.group import Group
+from sentry.models.options.project_option import ProjectOption
 from sentry.models.organization import Organization
 from sentry.models.project import Project
 from sentry.models.repository import Repository
 from sentry.net.http import connection_from_url
+from sentry.projectoptions.defaults import SEER_PROJECT_PREFERENCE_OPTION_KEYS
 from sentry.seer.autofix.constants import AutofixAutomationTuningSettings, AutofixStatus
 from sentry.seer.constants import SEER_SUPPORTED_SCM_PROVIDERS
 from sentry.seer.models import (
+    AutofixHandoffPoint,
     BranchOverride,
     SeerApiError,
     SeerApiResponseValidationError,
+    SeerAutomationHandoffConfiguration,
     SeerPermissionError,
     SeerProjectPreference,
     SeerRawPreferenceResponse,
@@ -44,10 +50,6 @@
     SeerProjectRepository,
     SeerProjectRepositoryBranchOverride,
 )
-from sentry.seer.models.seer_api_models import (
-    AutofixHandoffPoint,
-    SeerAutomationHandoffConfiguration,
-)
 from sentry.seer.signed_seer_api import SeerViewerContext, make_signed_seer_api_request
 from sentry.utils.cache import cache
 from sentry.utils.outcomes import Outcome, track_outcome
@@ -674,6 +676,145 @@ def bulk_write_preferences_to_sentry_db(
     _write_preferences_to_sentry_db(project_preferences)
 
 
+def build_repo_definition_from_project_repo(
+    seer_project_repo: SeerProjectRepository,
+) -> SeerRepoDefinition | None:
+    """Build a SeerRepoDefinition from a SeerProjectRepository with its joined Repository.
+
+    Returns None if Repository name is invalid."""
+    repo = seer_project_repo.repository
+    repo_name_sections = repo.name.split("/")
+    if len(repo_name_sections) < 2:
+        sentry_sdk.capture_exception(ValueError(f"Invalid repository name format: {repo.name}"))
+        return None
+
+    return SeerRepoDefinition(
+        repository_id=repo.id,
+        organization_id=repo.organization_id,
+        integration_id=str(repo.integration_id) if repo.integration_id is not None else None,
+        provider=repo.provider or "",
+        owner=repo_name_sections[0],
+        name="/".join(repo_name_sections[1:]),
+        external_id=repo.external_id or "",
+        branch_name=seer_project_repo.branch_name,
+        instructions=seer_project_repo.instructions,
+        branch_overrides=[
+            BranchOverride(
+                tag_name=bo.tag_name,
+                tag_value=bo.tag_value,
+                branch_name=bo.branch_name,
+            )
+            for bo in seer_project_repo.branch_overrides.all()
+        ],
+    )
+
+
+def _build_automation_handoff(
+    get_option: Callable[[str], Any],
+) -> SeerAutomationHandoffConfiguration | None:
+    """Build a SeerAutomationHandoffConfiguration from option values, or None if incomplete."""
+    handoff_point = get_option("sentry:seer_automation_handoff_point")
+    handoff_target = get_option("sentry:seer_automation_handoff_target")
+    handoff_integration_id = get_option("sentry:seer_automation_handoff_integration_id")
+
+    if handoff_point is None or handoff_target is None or handoff_integration_id is None:
+        return None
+
+    return SeerAutomationHandoffConfiguration(
+        handoff_point=handoff_point,
+        target=handoff_target,
+        integration_id=handoff_integration_id,
+        auto_create_pr=get_option("sentry:seer_automation_handoff_auto_create_pr"),
+    )
+
+
+def read_preference_from_sentry_db(project: Project) -> SeerProjectPreference | None:
+    """Read a single project's Seer preferences from Sentry DB.
+
+    For now, should only be used under feature flag `organizations:seer-project-settings-read-from-sentry`."""
+    seer_project_repo_qs = (
+        SeerProjectRepository.objects.filter(project=project)
+        .select_related("repository")
+        .prefetch_related("branch_overrides")
+    )
+    repo_definitions = [
+        repo_def
+        for project_repo in seer_project_repo_qs
+        if (repo_def := build_repo_definition_from_project_repo(project_repo)) is not None
+    ]
+
+    has_configured_options = any(
+        ProjectOption.objects.isset(project, key) for key in SEER_PROJECT_PREFERENCE_OPTION_KEYS
+    )
+    if not repo_definitions and not has_configured_options:
+        return None
+
+    return SeerProjectPreference(
+        organization_id=project.organization_id,
+        project_id=project.id,
+        repositories=repo_definitions,
+        automated_run_stopping_point=project.get_option("sentry:seer_automated_run_stopping_point"),
+        automation_handoff=_build_automation_handoff(project.get_option),
+    )
+
+
+def bulk_read_preferences_from_sentry_db(
+    organization_id: int, project_ids: list[int]
+) -> dict[int, SeerProjectPreference | None]:
+    """Bulk read Seer preferences from Sentry DB.
+
+    For now, should only be used under feature flag `organizations:seer-project-settings-read-from-sentry`."""
+    if not project_ids:
+        return {}
+
+    projects = list(Project.objects.filter(id__in=project_ids, organization_id=organization_id))
+
+    repo_definitions_by_project: defaultdict[int, list[SeerRepoDefinition]] = defaultdict(list)
+    for project_repo in (
+        SeerProjectRepository.objects.filter(project_id__in=project_ids)
+        .select_related("repository")
+        .prefetch_related("branch_overrides")
+    ):
+        repo_def = build_repo_definition_from_project_repo(project_repo)
+        if repo_def is not None:
+            repo_definitions_by_project[project_repo.project_id].append(repo_def)
+
+    # get_value_bulk_id returns None for missing options, unlike project.get_option
+    # which automatically falls back to the registered well-known key default.
+    project_options: dict[str, Mapping[int, Any]] = {
+        key: ProjectOption.objects.get_value_bulk_id(project_ids, key)
+        for key in SEER_PROJECT_PREFERENCE_OPTION_KEYS
+    }
+
+    result: dict[int, SeerProjectPreference | None] = {}
+    for project in projects:
+        has_configured_options = any(
+            project_options[key][project.id] is not None
+            for key in SEER_PROJECT_PREFERENCE_OPTION_KEYS
+        )
+        if project.id not in repo_definitions_by_project and not has_configured_options:
+            result[project.id] = None
+            continue
+
+        def _get_project_option(key: str) -> Any:
+            value = project_options[key][project.id]
+            if value is None:
+                return projectoptions.get_well_known_default(key, project=project)
+            return value
+
+        result[project.id] = SeerProjectPreference(
+            organization_id=project.organization_id,
+            project_id=project.id,
+            repositories=repo_definitions_by_project.get(project.id, []),
+            automated_run_stopping_point=_get_project_option(
+                "sentry:seer_automated_run_stopping_point"
+            ),
+            automation_handoff=_build_automation_handoff(_get_project_option),
+        )
+
+    return result
+
+
 def set_project_seer_preference(preference: SeerProjectPreference) -> None:
     """Set Seer project preference for a single project via Seer API."""
     response = make_set_project_preference_request(