feat(search): Downrank capture_message issues in recommended sort

src/sentry/options/defaults.py

@@ -901,6 +901,11 @@
     default={7001: 0.15},
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "snuba.search.recommended.message-penalty-weight",
+    default=0.10,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 
 # The percentage of tagkeys that we want to cache. Set to 1.0 in order to cache everything, <=0.0 to stop caching
 register(

src/sentry/search/snuba/executors.py

@@ -813,16 +813,27 @@ def _recommended_aggregation(
     else:
         type_boost = "0.0"
 
+    # Message penalty: downrank issues from capture_message (no exception/stacktrace)
+    # Only applies to Events dataset — issue-platform occurrences don't have exception_stacks
+    if type_column is None:
+        message_penalty_weight = options.get("snuba.search.recommended.message-penalty-weight")
+        has_exception_ratio = "divide(countIf(notEmpty(exception_stacks.type)), plus(count(), 1))"
+        message_penalty = f"multiply({message_penalty_weight}, minus(1.0, {has_exception_ratio}))"
+    else:
+        message_penalty = "0.0"
+
+    weighted_score = (
+        f"plus(plus(plus(plus(plus("
+        f"multiply({recency_weight}, {recency}), "
+        f"multiply({spike_weight}, {spike})), "
+        f"multiply({severity_weight}, {severity})), "
+        f"multiply({user_impact_weight}, {user_impact})), "
+        f"multiply({event_volume_weight}, {event_volume})), "
+        f"{type_boost})"
+    )
+
     return [
-        (
-            f"plus(plus(plus(plus(plus("
-            f"multiply({recency_weight}, {recency}), "
-            f"multiply({spike_weight}, {spike})), "
-            f"multiply({severity_weight}, {severity})), "
-            f"multiply({user_impact_weight}, {user_impact})), "
-            f"multiply({event_volume_weight}, {event_volume})), "
-            f"{type_boost})"
-        ),
+        f"minus({weighted_score}, {message_penalty})",
         "",
     ]
 

tests/snuba/search/test_backend.py

@@ -4122,3 +4122,57 @@ def test_recommended_group_type_boost(self) -> None:
 
         scores = {gid: score for gid, score in results}
         assert scores[profile_group.id] > scores[error_group.id]
+
+    def test_recommended_message_penalty(self) -> None:
+        base_datetime = before_now(hours=1)
+
+        error_event = self.store_event(
+            data={
+                "fingerprint": ["exception-group"],
+                "event_id": "a" * 32,
+                "timestamp": base_datetime.isoformat(),
+                "message": "real error",
+                "level": "error",
+                "exception": {
+                    "values": [
+                        {
+                            "type": "ValueError",
+                            "value": "something broke",
+                            "stacktrace": {"frames": [{"module": "app.main"}]},
+                        }
+                    ]
+                },
+                "tags": {"sentry:user": "user1@example.com"},
+            },
+            project_id=self.project.id,
+        )
+        exception_group = error_event.group
+
+        message_event = self.store_event(
+            data={
+                "fingerprint": ["message-group"],
+                "event_id": "b" * 32,
+                "timestamp": base_datetime.isoformat(),
+                "message": "disk space low",
+                "level": "error",
+                "tags": {"sentry:user": "user2@example.com"},
+            },
+            project_id=self.project.id,
+        )
+        message_group = message_event.group
+
+        query_executor = self.backend._get_query_executor()
+        results = query_executor.snuba_search(
+            start=None,
+            end=None,
+            project_ids=[self.project.id],
+            environment_ids=[],
+            sort_field="recommended",
+            organization=self.organization,
+            group_ids=[exception_group.id, message_group.id],
+            limit=150,
+            referrer=Referrer.TESTING_TEST,
+        )[0]
+
+        scores = {gid: score for gid, score in results}
+        assert scores[exception_group.id] > scores[message_group.id]