feat(triage signals): Use new seat based pricing feature flag

src/sentry/seer/autofix/issue_summary.py

@@ -27,6 +27,7 @@
     AutofixStoppingPoint,
     get_autofix_state,
     is_seer_autotriggered_autofix_rate_limited,
+    is_seer_seat_based_tier_enabled,
 )
 from sentry.seer.models import SummarizeIssueResponse
 from sentry.seer.seer_setup import get_seer_org_acknowledgement
@@ -341,8 +342,16 @@ def run_automation(
             )
             return
 
-    # Only log for orgs with triage-signals-v0-org
-    if features.has("organizations:triage-signals-v0-org", group.organization):
+    # Only log for orgs with seat-based Seer tier
+    try:
+        should_log = is_seer_seat_based_tier_enabled(group.organization)
+    except Exception:
+        logger.exception(
+            "Error checking if seat-based Seer tier is enabled", extra={"group_id": group.id}
+        )
+        should_log = False
+
+    if should_log:
         try:
             times_seen = group.times_seen_with_pending
         except (AssertionError, AttributeError):

src/sentry/seer/autofix/utils.py

@@ -28,6 +28,7 @@
 )
 from sentry.seer.signed_seer_api import make_signed_seer_api_request, sign_with_seer_secret
 from sentry.utils import json
+from sentry.utils.cache import cache
 from sentry.utils.outcomes import Outcome, track_outcome
 
 logger = logging.getLogger(__name__)
@@ -364,6 +365,30 @@ def is_seer_scanner_rate_limited(project: Project, organization: Organization) -
     return is_rate_limited
 
 
+def get_seer_seat_based_tier_cache_key(organization_id: int) -> str:
+    """Get the cache key for seat-based Seer tier check."""
+    return f"seer:seat-based-tier:{organization_id}"
+
+
+def is_seer_seat_based_tier_enabled(organization: Organization) -> bool:
+    """
+    Check if organization has Seer seat-based pricing via billing.
+    """
+    if features.has("organizations:triage-signals-v0-org", organization):
+        return True
+
+    cache_key = get_seer_seat_based_tier_cache_key(organization.id)
+    cached_value = cache.get(cache_key)
+    if cached_value is not None:
+        return cached_value
+
+    logger.info("Checking if seat-based Seer tier is enabled for organization=%s", organization.id)
+    has_seat_based_seer = features.has("organizations:seat-based-seer-enabled", organization)
+    cache.set(cache_key, has_seat_based_seer, timeout=60 * 60 * 4)  # 4 hours TTL
+
+    return has_seat_based_seer
+
+
 def is_issue_eligible_for_seer_automation(group: Group) -> bool:
     """Check if Seer automation is allowed for a given group based on permissions and issue type."""
     from sentry import quotas

src/sentry/tasks/autofix.py

@@ -9,10 +9,12 @@
     bulk_get_project_preferences,
     bulk_set_project_preferences,
     get_autofix_state,
+    get_seer_seat_based_tier_cache_key,
 )
 from sentry.tasks.base import instrumented_task
 from sentry.taskworker.namespaces import ingest_errors_tasks, issues_tasks
 from sentry.taskworker.retry import Retry
+from sentry.utils.cache import cache
 
 logger = logging.getLogger(__name__)
 
@@ -123,6 +125,9 @@ def configure_seer_for_existing_org(organization_id: int) -> None:
     # Set org-level options
     organization.update_option("sentry:enable_seer_coding", True)
 
+    # Invalidate seat-based tier cache so new settings take effect immediately
+    cache.delete(get_seer_seat_based_tier_cache_key(organization_id))
+
     projects = list(Project.objects.filter(organization_id=organization_id, status=0))
     project_ids = [p.id for p in projects]
 

tests/sentry/seer/autofix/test_autofix_utils.py

@@ -12,9 +12,11 @@
     get_autofix_prompt,
     get_coding_agent_prompt,
     is_issue_eligible_for_seer_automation,
+    is_seer_seat_based_tier_enabled,
 )
 from sentry.seer.models import SeerApiError
 from sentry.testutils.cases import TestCase
+from sentry.utils.cache import cache
 
 
 class TestGetAutofixPrompt(TestCase):
@@ -347,3 +349,57 @@ def test_returns_true_when_issue_type_always_triggers(
                 result = is_issue_eligible_for_seer_automation(self.group)
 
                 assert result is True
+
+
+class TestIsSeerSeatBasedTierEnabled(TestCase):
+    """Test the is_seer_seat_based_tier_enabled function."""
+
+    def setUp(self):
+        super().setUp()
+        self.organization = self.create_organization(name="test-org")
+
+    def tearDown(self):
+        super().tearDown()
+        cache.delete(f"seer:seat-based-tier:{self.organization.id}")
+
+    def test_returns_true_when_triage_signals_enabled(self):
+        """Test returns True when triage-signals-v0-org feature flag is enabled."""
+        with self.feature("organizations:triage-signals-v0-org"):
+            result = is_seer_seat_based_tier_enabled(self.organization)
+            assert result is True
+
+    @patch("sentry.seer.autofix.utils.features.has")
+    def test_returns_true_when_seat_based_seer_enabled(self, mock_features_has):
+        """Test returns True when seat-based-seer-enabled feature flag is enabled and caches the result."""
+
+        def features_side_effect(flag, org):
+            if flag == "organizations:seat-based-seer-enabled":
+                return True
+            return False
+
+        mock_features_has.side_effect = features_side_effect
+
+        result = is_seer_seat_based_tier_enabled(self.organization)
+        assert result is True
+
+        # Verify it was cached
+        cache_key = f"seer:seat-based-tier:{self.organization.id}"
+        assert cache.get(cache_key) is True
+
+    def test_returns_false_when_no_flags_enabled(self):
+        """Test returns False when neither feature flag is enabled and caches the result."""
+        result = is_seer_seat_based_tier_enabled(self.organization)
+        assert result is False
+
+        # Verify False was cached
+        cache_key = f"seer:seat-based-tier:{self.organization.id}"
+        assert cache.get(cache_key) is False
+
+    def test_returns_cached_value(self):
+        """Test returns cached value without checking feature flags."""
+        cache_key = f"seer:seat-based-tier:{self.organization.id}"
+        cache.set(cache_key, True, timeout=60)
+
+        # Even without feature flags enabled, should return cached True
+        result = is_seer_seat_based_tier_enabled(self.organization)
+        assert result is True

tests/sentry/tasks/test_autofix.py

@@ -5,14 +5,15 @@
 from django.test import TestCase
 
 from sentry.seer.autofix.constants import AutofixStatus, SeerAutomationSource
-from sentry.seer.autofix.utils import AutofixState
+from sentry.seer.autofix.utils import AutofixState, get_seer_seat_based_tier_cache_key
 from sentry.seer.models import SeerApiError, SummarizeIssueResponse, SummarizeIssueScores
 from sentry.tasks.autofix import (
     check_autofix_status,
     configure_seer_for_existing_org,
     generate_issue_summary_only,
 )
 from sentry.testutils.cases import TestCase as SentryTestCase
+from sentry.utils.cache import cache
 
 
 class TestCheckAutofixStatus(TestCase):
@@ -233,3 +234,22 @@ def test_raises_on_bulk_set_api_failure(
         # Sentry DB options should still be set before the API call
         assert project1.get_option("sentry:seer_scanner_automation") is True
         assert project2.get_option("sentry:seer_scanner_automation") is True
+
+    @patch("sentry.tasks.autofix.bulk_set_project_preferences")
+    @patch("sentry.tasks.autofix.bulk_get_project_preferences")
+    def test_invalidates_seat_based_tier_cache(
+        self, mock_bulk_get: MagicMock, mock_bulk_set: MagicMock
+    ) -> None:
+        """Test that the seat-based tier cache is invalidated after configuring org."""
+        self.create_project(organization=self.organization)
+        mock_bulk_get.return_value = {}
+
+        # Set a cached value before running the task
+        cache_key = get_seer_seat_based_tier_cache_key(self.organization.id)
+        cache.set(cache_key, False, timeout=60 * 60 * 4)
+        assert cache.get(cache_key) is False
+
+        configure_seer_for_existing_org(organization_id=self.organization.id)
+
+        # Cache should be invalidated
+        assert cache.get(cache_key) is None