feat(search): Add issue progress sort

src/sentry/features/temporary.py

@@ -326,6 +326,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:issue-stream-progress-ui", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable the experimental "recommended" sort option in the issue stream
     manager.add("organizations:issue-stream-recommended-sort", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    # Enable the "progress" sort option (by issue fix-cycle progress) in the issue stream
+    manager.add("organizations:issue-stream-progress-sort", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
 
     # Lets organizations manage grouping configs
     manager.add("organizations:set-grouping-config", OrganizationFeature, FeatureHandlerStrategy.INTERNAL, api_expose=True)

src/sentry/issues/endpoints/organization_group_index.py

@@ -12,7 +12,7 @@
 from rest_framework.response import Response
 from sentry_sdk import start_span
 
-from sentry import analytics, search
+from sentry import analytics, features, search
 from sentry.analytics.events.issue_search_endpoint_queried import IssueSearchEndpointQueriedEvent
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
@@ -54,7 +54,7 @@
 )
 from sentry.apidocs.response_types import DetailResponse, ValidationErrorResponse
 from sentry.apidocs.utils import inline_sentry_response_serializer
-from sentry.constants import ALLOWED_FUTURE_DELTA
+from sentry.constants import ALLOWED_FUTURE_DELTA, DEFAULT_SORT_OPTION
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.environment import Environment
 from sentry.models.group import QUERY_STATUS_LOOKUP, Group, GroupStatus
@@ -183,6 +183,10 @@ def search_issues(
         query_kwargs["environments"] = environments if environments else None
 
         query_kwargs["actor"] = request.user
+        if query_kwargs["sort_by"] == "progress" and not features.has(
+            "organizations:issue-stream-progress-sort", organization, actor=request.user
+        ):
+            query_kwargs["sort_by"] = DEFAULT_SORT_OPTION
         if query_kwargs["sort_by"] == "inbox":
             query_kwargs.pop("sort_by")
             query_kwargs.pop("referrer")

src/sentry/search/snuba/executors.py

@@ -24,6 +24,7 @@
 from sentry.constants import ALLOWED_FUTURE_DELTA
 from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.issues.grouptype import GroupCategory
+from sentry.issues.progress import IssueProgressState, get_group_progress_states
 from sentry.issues.search import (
     SEARCH_FILTER_UPDATERS,
     IntermediateSearchQueryPartial,
@@ -986,6 +987,54 @@ def score_fn(data: dict[str, Any]) -> float:
     )
 
 
+# Numeric rank for the "progress" sort: higher means further along the fix cycle, so it
+# sorts towards the top. Every state has a rank so issues without seer activity (the
+# identified/triaged base states) still order correctly relative to progressed issues.
+PROGRESS_STATE_SORT_RANK: dict[IssueProgressState, int] = {
+    IssueProgressState.IDENTIFIED: 1,
+    IssueProgressState.TRIAGED: 2,
+    IssueProgressState.DIAGNOSED: 3,
+    IssueProgressState.FIX_PROPOSED: 4,
+    IssueProgressState.FIX_APPLIED: 5,
+}
+
+# last_seen comes back from Snuba as epoch milliseconds (< 1e13 until the year 2286), so
+# dividing by this collapses it into a [0, 1) recency fraction. The score is then
+# `rank + fraction`: rank stays the primary (integer) key and last_seen only breaks ties.
+LAST_SEEN_TIEBREAK_DIVISOR = 10**13
+
+
+def resolve_progress_signal(
+    actor: Any | None, organization: Organization, group_ids: list[int]
+) -> dict[int, int]:
+    """Progress-cycle rank per group (identified=1 .. fix_applied=5), derived from the same
+    Activity records as the ``issue.progress`` filter. Every group gets a rank."""
+    states = get_group_progress_states(group_ids)
+    return {
+        group_id: PROGRESS_STATE_SORT_RANK[IssueProgressState(state)]
+        for group_id, state in states.items()
+    }
+
+
+def progress_strategy() -> PostgresSortStrategy:
+    """Progress sort: primary by fix-cycle rank (fix_applied > fix_proposed > diagnosed >
+    triaged > identified), secondary by last_seen. The secondary key stands in for
+    ``issue.last_progressed_at`` until that field exists; for now most-recently-active issues
+    rank highest within a tier."""
+
+    def score_fn(data: dict[str, Any]) -> float:
+        rank = data.get("progress_rank") or 0
+        last_seen = data.get("last_seen") or 0
+        return rank + last_seen / LAST_SEEN_TIEBREAK_DIVISOR
+
+    return PostgresSortStrategy(
+        postgres_fields={},
+        snuba_aggregations=["last_seen"],
+        signal_resolvers={"progress_rank": resolve_progress_signal},
+        score_fn=score_fn,
+    )
+
+
 class PostgresSnubaQueryExecutor(AbstractQueryExecutor):
     ISSUE_FIELD_NAME = "group_id"
 
@@ -1006,6 +1055,10 @@ class PostgresSnubaQueryExecutor(AbstractQueryExecutor):
         # Snuba path can take over when there are too many candidates to score in memory.
         "recommended_v2": "recommended",
         "user": "user_count",
+        # Postgres-data sort; mapped to last_seen so the chunked Snuba path can take over
+        # (degrading to a plain last_seen sort) when there are too many candidates to score
+        # the progress rank in memory.
+        "progress": "last_seen",
         # We don't need a corresponding snuba field here, since this sort only happens
         # in Postgres
         "inbox": "",
@@ -1030,7 +1083,10 @@ def dataset(self) -> Dataset:
 
     @property
     def postgres_sort_strategies(self) -> dict[str, PostgresSortStrategy]:
-        return {"recommended_v2": recommended_v2_strategy()}
+        return {
+            "recommended_v2": recommended_v2_strategy(),
+            "progress": progress_strategy(),
+        }
 
     def _apply_type_visibility_filter(
         self,

tests/sentry/issues/endpoints/test_organization_group_index.py

@@ -190,6 +190,29 @@ def test_sort_by_trends(self) -> None:
         assert len(response.data) == 2
         assert [item["id"] for item in response.data] == [str(group.id), str(group_2.id)]
 
+    def test_sort_by_progress_requires_feature_flag(self) -> None:
+        # group_1 has the newer event (wins last_seen / default sort); group_2 is older but
+        # diagnosed, so it wins the progress sort once the flag is on.
+        group_1 = self.store_event(
+            data={"timestamp": before_now(seconds=1).isoformat(), "fingerprint": ["group-1"]},
+            project_id=self.project.id,
+        ).group
+        group_2 = self.store_event(
+            data={"timestamp": before_now(hours=1).isoformat(), "fingerprint": ["group-2"]},
+            project_id=self.project.id,
+        ).group
+        self.create_group_activity(group=group_2, type=ActivityType.SEER_RCA_COMPLETED.value)
+        self.login_as(user=self.user)
+
+        # Without the flag, the sort falls back to the default (date) order.
+        response = self.get_success_response(sort="progress", query="is:unresolved")
+        assert [item["id"] for item in response.data] == [str(group_1.id), str(group_2.id)]
+
+        # With the flag, the diagnosed group is promoted above the more recently seen one.
+        with self.feature("organizations:issue-stream-progress-sort"):
+            response = self.get_success_response(sort="progress", query="is:unresolved")
+        assert [item["id"] for item in response.data] == [str(group_2.id), str(group_1.id)]
+
     def test_sort_by_inbox(self) -> None:
         group_1 = self.store_event(
             data={

tests/snuba/search/test_postgres_sort_framework.py

@@ -455,11 +455,59 @@ def test_agent_boost_reset_by_regression(self):
         assert self._query(actor=self.user) == [self.groups[1], self.groups[2], self.groups[0]]
 
 
+class TestProgressSort(PostgresSortTestBase):
+    """progress: primary sort by fix-cycle rank (fix_applied > fix_proposed > diagnosed >
+    triaged > identified), secondary by last_seen.
+
+    The base fixture's groups have events ~8d, ~5d, and ~3d old, so on last_seen alone they
+    order [2, 1, 0] (newest first).
+    """
+
+    def _query(self):
+        return list(
+            self.backend.query(
+                [self.project],
+                search_filters=[],
+                environments=None,
+                count_hits=False,
+                sort_by="progress",
+                date_from=None,
+                date_to=None,
+                cursor=None,
+                referrer=Referrer.TESTING_TEST,
+            )
+        )
+
+    def test_rank_outranks_last_seen(self):
+        # Give the oldest group the furthest progress and the newest group none: rank must
+        # invert the last_seen ordering.
+        self.create_group_activity(group=self.groups[0], type=ActivityType.SEER_PR_CREATED.value)
+        self.create_group_activity(group=self.groups[1], type=ActivityType.SEER_RCA_COMPLETED.value)
+        # groups[2] has no progress activity -> identified (lowest rank).
+        assert self._query() == [self.groups[0], self.groups[1], self.groups[2]]
+
+    def test_last_seen_breaks_ties_within_rank(self):
+        # groups[0] and groups[1] are both diagnosed; the more recently seen (groups[1])
+        # sorts first. groups[2] stays identified and sorts last.
+        self.create_group_activity(group=self.groups[0], type=ActivityType.SEER_RCA_COMPLETED.value)
+        self.create_group_activity(group=self.groups[1], type=ActivityType.SEER_RCA_COMPLETED.value)
+        assert self._query() == [self.groups[1], self.groups[0], self.groups[2]]
+
+
 class TestDefaultPostgresSortStrategies(TestCase):
     def test_recommended_v2_registered(self):
         strategies = PostgresSnubaQueryExecutor().postgres_sort_strategies
-        assert set(strategies) == {"recommended_v2"}
+        assert set(strategies) == {"recommended_v2", "progress"}
         strategy = strategies["recommended_v2"]
         assert strategy.snuba_aggregations == ["recommended"]
         assert strategy.exclude_null_postgres is False
         assert set(strategy.signal_resolvers) == {"assignment", "agent"}
+
+    def test_progress_registered(self):
+        strategies = PostgresSnubaQueryExecutor().postgres_sort_strategies
+        strategy = strategies["progress"]
+        assert strategy.snuba_aggregations == ["last_seen"]
+        assert set(strategy.signal_resolvers) == {"progress_rank"}
+        # progress maps to last_seen in sort_strategies so the chunked Snuba path has a
+        # real aggregation to fall back to on candidate overflow.
+        assert PostgresSnubaQueryExecutor.sort_strategies["progress"] == "last_seen"