feat(triage signals): New kick_off_seer_automation flow [feature flagged] (#103676)

## PR Details
+ I had to revert by last fixability
[PR](#103485). I am going to
address it separately after this PR - made a ticket for that:
https://linear.app/getsentry/issue/AIML-1650/address-fixability-issue-in-the-new-automation-flow
  + Also mentioned in the TODO comment
+ This PR just creates the new automation flow for triage signals behind
a feature flag.
+ Reference diagram:
https://miro.com/app/board/uXjVJqn1-fQ=/?focusWidget=3458764648325594380
+ Also make `run_automation` a public method now since we directly use
in a task.

Author: Mihir-Mavalankar

src/sentry/seer/autofix/issue_summary.py

@@ -290,7 +290,7 @@ def _is_issue_fixable(group: Group, fixability_score: float) -> bool:
     return False
 
 
-def _run_automation(
+def run_automation(
     group: Group,
     user: User | RpcUser | AnonymousUser,
     event: GroupEvent,
@@ -402,7 +402,7 @@ def _generate_summary(
 
     if should_run_automation:
         try:
-            _run_automation(group, user, event, source)
+            run_automation(group, user, event, source)
         except Exception:
             logger.exception(
                 "Error auto-triggering autofix from issue summary", extra={"group_id": group.id}

src/sentry/tasks/autofix.py

@@ -31,12 +31,12 @@ def check_autofix_status(run_id: int, organization_id: int) -> None:
 
 
 @instrumented_task(
-    name="sentry.tasks.autofix.start_seer_automation",
+    name="sentry.tasks.autofix.generate_summary_and_run_automation",
     namespace=ingest_errors_tasks,
     processing_deadline_duration=35,
     retry=Retry(times=1),
 )
-def start_seer_automation(group_id: int) -> None:
+def generate_summary_and_run_automation(group_id: int) -> None:
     from sentry.seer.autofix.issue_summary import get_issue_summary
 
     group = Group.objects.get(id=group_id)
@@ -60,3 +60,33 @@ def generate_issue_summary_only(group_id: int) -> None:
     get_issue_summary(
         group=group, source=SeerAutomationSource.POST_PROCESS, should_run_automation=False
     )
+    # TODO: Generate fixability score here and check for it in run_automation around line 316
+    # That will make sure that even after adding fixability here it's not re-triggered.
+    # Currently fixability will only be generated after 10 events when run_automation is called
+
+
+@instrumented_task(
+    name="sentry.tasks.autofix.run_automation_only_task",
+    namespace=ingest_errors_tasks,
+    processing_deadline_duration=35,
+    retry=Retry(times=1),
+)
+def run_automation_only_task(group_id: int) -> None:
+    """
+    Run automation directly for a group (assumes summary and fixability already exist).
+    Used for triage signals flow when event count >= 10 and summary exists.
+    """
+    from django.contrib.auth.models import AnonymousUser
+
+    from sentry.seer.autofix.issue_summary import run_automation
+
+    group = Group.objects.get(id=group_id)
+    event = group.get_latest_event()
+
+    if not event:
+        logger.warning("run_automation_only_task.no_event_found", extra={"group_id": group_id})
+        return
+
+    run_automation(
+        group=group, user=AnonymousUser(), event=event, source=SeerAutomationSource.POST_PROCESS
+    )

src/sentry/tasks/post_process.py

@@ -1595,33 +1595,100 @@ def check_if_flags_sent(job: PostProcessJob) -> None:
 
 
 def kick_off_seer_automation(job: PostProcessJob) -> None:
-    from sentry.seer.autofix.issue_summary import get_issue_summary_lock_key
+    from sentry.seer.autofix.constants import AutofixAutomationTuningSettings
+    from sentry.seer.autofix.issue_summary import (
+        get_issue_summary_cache_key,
+        get_issue_summary_lock_key,
+    )
     from sentry.seer.autofix.utils import (
         is_issue_eligible_for_seer_automation,
         is_seer_scanner_rate_limited,
     )
-    from sentry.tasks.autofix import start_seer_automation
+    from sentry.tasks.autofix import (
+        generate_issue_summary_only,
+        generate_summary_and_run_automation,
+        run_automation_only_task,
+    )
 
     event = job["event"]
     group = event.group
 
-    # Only run on issues with no existing scan - TODO: Update condition for triage signals V0
-    if group.seer_fixability_score is not None:
-        return
+    # Default behaviour
+    if not features.has("projects:triage-signals-v0", group.project):
+        # Only run on issues with no existing scan
+        if group.seer_fixability_score is not None:
+            return
 
-    if is_issue_eligible_for_seer_automation(group) is False:
-        return
+        if not is_issue_eligible_for_seer_automation(group):
+            return
 
-    # Don't run if there's already a task in progress for this issue
-    lock_key, lock_name = get_issue_summary_lock_key(group.id)
-    lock = locks.get(lock_key, duration=1, name=lock_name)
-    if lock.locked():
-        return
+        # Don't run if there's already a task in progress for this issue
+        lock_key, lock_name = get_issue_summary_lock_key(group.id)
+        lock = locks.get(lock_key, duration=1, name=lock_name)
+        if lock.locked():
+            return
 
-    if is_seer_scanner_rate_limited(group.project, group.organization):
-        return
+        if is_seer_scanner_rate_limited(group.project, group.organization):
+            return
+
+        generate_summary_and_run_automation.delay(group.id)
+    else:
+        # Triage signals V0 behaviour
+
+        # If event count < 10, only generate summary (no automation)
+        if group.times_seen_with_pending < 10:
+            # Check if summary exists in cache
+            cache_key = get_issue_summary_cache_key(group.id)
+            if cache.get(cache_key) is not None:
+                return
+
+            # Early returns for eligibility checks (cheap checks first)
+            if not is_issue_eligible_for_seer_automation(group):
+                return
+
+            # Atomically set cache to prevent duplicate summary generation
+            summary_dispatch_cache_key = f"seer-summary-dispatched:{group.id}"
+            if not cache.add(summary_dispatch_cache_key, True, timeout=30):
+                return  # Another process already dispatched summary generation
+
+            # Rate limit check must be last, after cache.add succeeds, to avoid wasting quota
+            if is_seer_scanner_rate_limited(group.project, group.organization):
+                return
+
+            generate_issue_summary_only.delay(group.id)
+        else:
+            # Event count >= 10: run automation
+            # Long-term check to avoid re-running
+            if (
+                group.seer_autofix_last_triggered is not None
+                or group.seer_fixability_score
+                is not None  # TODO: Remove this once fixability is generated with generate_issue_summary_only
+                or group.project.get_option("sentry:autofix_automation_tuning")
+                == AutofixAutomationTuningSettings.OFF
+            ):
+                return
+
+            # Early returns for eligibility checks (cheap checks first)
+            if not is_issue_eligible_for_seer_automation(group):
+                return
 
-    start_seer_automation.delay(group.id)
+            # Atomically set cache to prevent duplicate dispatches (returns False if key exists)
+            automation_dispatch_cache_key = f"seer-automation-dispatched:{group.id}"
+            if not cache.add(automation_dispatch_cache_key, True, timeout=300):
+                return  # Another process already dispatched automation
+
+            # Check if summary exists in cache
+            cache_key = get_issue_summary_cache_key(group.id)
+            if cache.get(cache_key) is not None:
+                # Summary exists, run automation directly
+                run_automation_only_task.delay(group.id)
+            else:
+                # Rate limit check before generating summary
+                if is_seer_scanner_rate_limited(group.project, group.organization):
+                    return
+
+                # No summary yet, generate summary + run automation in one go
+                generate_summary_and_run_automation.delay(group.id)
 
 
 GROUP_CATEGORY_POST_PROCESS_PIPELINE = {

tests/sentry/seer/autofix/test_issue_summary.py

@@ -18,8 +18,8 @@
     _fetch_user_preference,
     _get_event,
     _get_stopping_point_from_fixability,
-    _run_automation,
     get_issue_summary,
+    run_automation,
 )
 from sentry.seer.autofix.utils import AutofixStoppingPoint
 from sentry.seer.models import SummarizeIssueResponse, SummarizeIssueScores
@@ -611,7 +611,7 @@ def test_get_issue_summary_with_web_vitals_issue(
         mock_trigger_autofix_task.assert_called_once()
 
     @patch("sentry.seer.autofix.issue_summary.get_seer_org_acknowledgement")
-    @patch("sentry.seer.autofix.issue_summary._run_automation")
+    @patch("sentry.seer.autofix.issue_summary.run_automation")
     @patch("sentry.seer.autofix.issue_summary._get_trace_tree_for_event")
     @patch("sentry.seer.autofix.issue_summary._call_seer")
     @patch("sentry.seer.autofix.issue_summary._get_event")
@@ -623,7 +623,7 @@ def test_get_issue_summary_continues_when_automation_fails(
         mock_run_automation,
         mock_get_acknowledgement,
     ):
-        """Test that issue summary is still returned when _run_automation throws an exception."""
+        """Test that issue summary is still returned when run_automation throws an exception."""
         mock_get_acknowledgement.return_value = True
 
         # Set up event and seer response
@@ -641,7 +641,7 @@ def test_get_issue_summary_continues_when_automation_fails(
         )
         mock_call_seer.return_value = mock_summary
 
-        # Make _run_automation raise an exception
+        # Make run_automation raise an exception
         mock_run_automation.side_effect = Exception("Automation failed")
 
         # Call get_issue_summary and verify it still returns successfully
@@ -652,7 +652,7 @@ def test_get_issue_summary_continues_when_automation_fails(
         expected_response["event_id"] = event.event_id
         assert summary_data == convert_dict_key_case(expected_response, snake_to_camel_case)
 
-        # Verify _run_automation was called and failed
+        # Verify run_automation was called and failed
         mock_run_automation.assert_called_once()
         mock_call_seer.assert_called_once()
 
@@ -681,7 +681,7 @@ def test_get_issue_summary_handles_trace_tree_errors(
                     possible_cause="cause",
                 ),
             ) as mock_call_seer,
-            patch("sentry.seer.autofix.issue_summary._run_automation"),
+            patch("sentry.seer.autofix.issue_summary.run_automation"),
             patch(
                 "sentry.seer.autofix.issue_summary.get_seer_org_acknowledgement",
                 return_value=True,
@@ -693,7 +693,7 @@ def test_get_issue_summary_handles_trace_tree_errors(
         mock_call_seer.assert_called_once_with(self.group, serialized_event, None)
 
     @patch("sentry.seer.autofix.issue_summary.get_seer_org_acknowledgement")
-    @patch("sentry.seer.autofix.issue_summary._run_automation")
+    @patch("sentry.seer.autofix.issue_summary.run_automation")
     @patch("sentry.seer.autofix.issue_summary._get_trace_tree_for_event")
     @patch("sentry.seer.autofix.issue_summary._call_seer")
     @patch("sentry.seer.autofix.issue_summary._get_event")
@@ -705,7 +705,7 @@ def test_get_issue_summary_with_should_run_automation_false(
         mock_run_automation,
         mock_get_acknowledgement,
     ):
-        """Test that should_run_automation=False prevents _run_automation from being called."""
+        """Test that should_run_automation=False prevents run_automation from being called."""
         mock_get_acknowledgement.return_value = True
         event = Mock(
             event_id="test_event_id",
@@ -743,7 +743,7 @@ def test_get_issue_summary_with_should_run_automation_false(
         mock_call_seer.assert_called_once_with(self.group, serialized_event, {"trace": "tree"})
         mock_get_acknowledgement.assert_called_once_with(self.group.organization)
 
-        # Verify that _run_automation was NOT called
+        # Verify that run_automation was NOT called
         mock_run_automation.assert_not_called()
 
         # Check if the cache was set correctly
@@ -798,7 +798,7 @@ def test_high_fixability_code_changes(
             possible_cause="c",
             scores=SummarizeIssueScores(fixability_score=0.70),
         )
-        _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+        run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
         mock_trigger.assert_called_once()
         assert mock_trigger.call_args[1]["stopping_point"] == AutofixStoppingPoint.CODE_CHANGES
 
@@ -822,7 +822,7 @@ def test_medium_fixability_solution(
             possible_cause="c",
             scores=SummarizeIssueScores(fixability_score=0.50),
         )
-        _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+        run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
         mock_trigger.assert_called_once()
         assert mock_trigger.call_args[1]["stopping_point"] == AutofixStoppingPoint.ROOT_CAUSE
 
@@ -848,7 +848,7 @@ def test_without_feature_flag(self, mock_gen, mock_budget, mock_state, mock_rate
         with self.feature(
             {"organizations:gen-ai-features": True, "projects:triage-signals-v0": False}
         ):
-            _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+            run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
 
         mock_trigger.assert_called_once()
         assert mock_trigger.call_args[1]["stopping_point"] is None
@@ -1001,7 +1001,7 @@ def test_user_preference_limits_high_fixability(
         )
         mock_fetch.return_value = "solution"
 
-        _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+        run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
 
         mock_trigger.assert_called_once()
         # Should be limited to SOLUTION by user preference
@@ -1031,7 +1031,7 @@ def test_fixability_limits_permissive_user_preference(
         )
         mock_fetch.return_value = "open_pr"
 
-        _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+        run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
 
         mock_trigger.assert_called_once()
         # Should use ROOT_CAUSE from fixability, not OPEN_PR from user
@@ -1061,7 +1061,7 @@ def test_no_user_preference_uses_fixability_only(
         )
         mock_fetch.return_value = None
 
-        _run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
+        run_automation(self.group, self.user, self.event, SeerAutomationSource.ALERT)
 
         mock_trigger.assert_called_once()
         # Should use OPEN_PR from fixability