fix(api): use hybrid aggregation path for finding-group filters

- Aggregate from findings when finding-level filters are present (region, delta, etc.)
- Keep summary path for compatible filters to preserve performance
- Apply status/severity filters post-aggregation in both paths
- Fix sort by time fields mapping to aggregated annotation names
- Use DenseRank instead of RowNumber to keep all findings per latest scan
- Expose muted_reason in finding-group resource endpoints
- Add tests for combined filters, validation errors, and time-field sorting
- Deduplicate list/latest actions via shared helpers
- Consolidate check_title icontains filter in base class

api/src/backend/api/filters.py

@@ -265,6 +265,13 @@ class CommonFindingFilters(FilterSet):
             )
         return queryset.filter(overall_query).distinct()
 
+    def filter_check_title_icontains(self, queryset, name, value):
+        return queryset.filter(
+            Q(check_metadata__CheckTitle__icontains=value)
+            | Q(check_metadata__checktitle__icontains=value)
+            | Q(check_metadata__Checktitle__icontains=value)
+        )
+
 
 class TenantFilter(FilterSet):
     inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
@@ -804,9 +811,7 @@ class FindingGroupFilter(CommonFindingFilters):
     check_id = CharFilter(field_name="check_id", lookup_expr="exact")
     check_id__in = CharInFilter(field_name="check_id", lookup_expr="in")
     check_id__icontains = CharFilter(field_name="check_id", lookup_expr="icontains")
-    check_title__icontains = CharFilter(
-        field_name="check_metadata__CheckTitle", lookup_expr="icontains"
-    )
+    check_title__icontains = CharFilter(method="filter_check_title_icontains")
     scan = UUIDFilter(field_name="scan_id", lookup_expr="exact")
     scan__in = UUIDInFilter(field_name="scan_id", lookup_expr="in")
 
@@ -902,9 +907,7 @@ class LatestFindingGroupFilter(CommonFindingFilters):
     check_id = CharFilter(field_name="check_id", lookup_expr="exact")
     check_id__in = CharInFilter(field_name="check_id", lookup_expr="in")
     check_id__icontains = CharFilter(field_name="check_id", lookup_expr="icontains")
-    check_title__icontains = CharFilter(
-        field_name="check_metadata__CheckTitle", lookup_expr="icontains"
-    )
+    check_title__icontains = CharFilter(method="filter_check_title_icontains")
     scan = UUIDFilter(field_name="scan_id", lookup_expr="exact")
     scan__in = UUIDInFilter(field_name="scan_id", lookup_expr="in")
 

api/src/backend/api/tests/test_views.py

@@ -45,7 +45,6 @@ from api.models import (
     ComplianceRequirementOverview,
     DailySeveritySummary,
     Finding,
-    FindingGroupDailySummary,
     Integration,
     Invitation,
     LighthouseProviderConfiguration,
@@ -15217,6 +15216,29 @@ class TestFindingGroupViewSet:
         # ec2_instance_public_ip has 1 PASS and 1 FAIL, should aggregate to FAIL
         assert data[0]["attributes"]["status"] == "FAIL"
 
+    def test_finding_groups_region_filter_reaggregates_metrics(
+        self, authenticated_client, finding_groups_fixture
+    ):
+        """Test finding-level filters recompute group metrics from matching findings."""
+        response = authenticated_client.get(
+            reverse("finding-group-list"),
+            {
+                "filter[inserted_at]": TODAY,
+                "filter[check_id]": "ec2_instance_public_ip",
+                "filter[region]": "us-east-1",
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+
+        attrs = data[0]["attributes"]
+        assert attrs["status"] == "PASS"
+        assert attrs["pass_count"] == 1
+        assert attrs["fail_count"] == 0
+        assert attrs["resources_total"] == 1
+        assert attrs["resources_fail"] == 0
+
     def test_finding_groups_status_pass_when_no_fail(
         self, authenticated_client, finding_groups_fixture
     ):
@@ -15284,6 +15306,103 @@ class TestFindingGroupViewSet:
         assert len(data) > 0
         assert all(item["attributes"]["severity"] == "critical" for item in data)
 
+    @pytest.mark.parametrize(
+        "endpoint_name", ["finding-group-list", "finding-group-latest"]
+    )
+    def test_finding_groups_combined_region_and_status_filters(
+        self, authenticated_client, finding_groups_fixture, endpoint_name
+    ):
+        """Test combined region + aggregated status filters."""
+        params = {"filter[region]": "us-east-1", "filter[status]": "FAIL"}
+        if endpoint_name == "finding-group-list":
+            params["filter[inserted_at]"] = TODAY
+
+        response = authenticated_client.get(reverse(endpoint_name), params)
+
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        check_ids = {item["id"] for item in data}
+        assert check_ids == {"s3_bucket_public_access", "cloudtrail_enabled"}
+        assert all(item["attributes"]["status"] == "FAIL" for item in data)
+
+    @pytest.mark.parametrize(
+        "endpoint_name", ["finding-group-list", "finding-group-latest"]
+    )
+    def test_finding_groups_combined_delta_and_severity_filters(
+        self, authenticated_client, finding_groups_fixture, endpoint_name
+    ):
+        """Test combined delta + aggregated severity filters."""
+        params = {"filter[delta]": "new", "filter[severity]": "critical"}
+        if endpoint_name == "finding-group-list":
+            params["filter[inserted_at]"] = TODAY
+
+        response = authenticated_client.get(reverse(endpoint_name), params)
+
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        check_ids = {item["id"] for item in data}
+        assert check_ids == {"s3_bucket_public_access", "cloudtrail_enabled"}
+        assert all(item["attributes"]["severity"] == "critical" for item in data)
+
+    @pytest.mark.parametrize(
+        "endpoint_name", ["finding-group-list", "finding-group-latest"]
+    )
+    @pytest.mark.parametrize(
+        "filter_key,filter_value",
+        [
+            ("status", "INVALID_STATUS"),
+            ("severity", "INVALID_SEVERITY"),
+        ],
+    )
+    def test_finding_groups_invalid_status_or_severity_returns_400(
+        self,
+        authenticated_client,
+        finding_groups_fixture,
+        endpoint_name,
+        filter_key,
+        filter_value,
+    ):
+        """Test invalid aggregated status/severity values are rejected."""
+        params = {f"filter[{filter_key}]": filter_value}
+        if endpoint_name == "finding-group-list":
+            params["filter[inserted_at]"] = TODAY
+
+        response = authenticated_client.get(reverse(endpoint_name), params)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json()["errors"][0]["code"] == "invalid"
+
+    @pytest.mark.parametrize(
+        "endpoint_name", ["finding-group-list", "finding-group-latest"]
+    )
+    @pytest.mark.parametrize(
+        "filter_key,filter_value,expected_detail",
+        [
+            ("status__in", "FAIL,INVALID_STATUS", "invalid status filter"),
+            ("severity__in", "critical,INVALID_SEVERITY", "invalid severity filter"),
+        ],
+    )
+    def test_finding_groups_invalid_in_filters_return_400(
+        self,
+        authenticated_client,
+        finding_groups_fixture,
+        endpoint_name,
+        filter_key,
+        filter_value,
+        expected_detail,
+    ):
+        """Test invalid values in status__in/severity__in are rejected."""
+        params = {f"filter[{filter_key}]": filter_value}
+        if endpoint_name == "finding-group-list":
+            params["filter[inserted_at]"] = TODAY
+
+        response = authenticated_client.get(reverse(endpoint_name), params)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        errors = response.json()["errors"]
+        assert errors[0]["code"] == "invalid"
+        assert expected_detail in errors[0]["detail"]
+
     @pytest.mark.parametrize(
         "filter_name,filter_value",
         [
@@ -15945,6 +16064,28 @@ class TestFindingGroupViewSet:
         assert len(data) > 0
         assert all(item["attributes"]["status"] == "FAIL" for item in data)
 
+    def test_finding_groups_latest_region_filter_reaggregates_metrics(
+        self, authenticated_client, finding_groups_fixture
+    ):
+        """Test /latest recomputes metrics from findings matching region filter."""
+        response = authenticated_client.get(
+            reverse("finding-group-latest"),
+            {
+                "filter[check_id]": "ec2_instance_public_ip",
+                "filter[region]": "us-east-1",
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+
+        attrs = data[0]["attributes"]
+        assert attrs["status"] == "PASS"
+        assert attrs["pass_count"] == 1
+        assert attrs["fail_count"] == 0
+        assert attrs["resources_total"] == 1
+        assert attrs["resources_fail"] == 0
+
     def test_finding_groups_latest_status_in_filter(
         self, authenticated_client, finding_groups_fixture
     ):
@@ -16043,46 +16184,124 @@ class TestFindingGroupViewSet:
         assert all(item["attributes"]["new_count"] > 0 for item in data)
 
     def test_finding_groups_latest_aggregates_latest_per_provider(
-        self, authenticated_client, providers_fixture
+        self,
+        authenticated_client,
+        providers_fixture,
+        resources_fixture,
     ):
-        """Test /latest aggregates latest summary from each provider for the same check."""
+        """Test /latest keeps all findings from the latest scan per provider.
+
+        Verifies that when the latest scan produces multiple findings for the
+        same check_id (e.g. one per resource), all of them are included in the
+        aggregation — not just one.
+        """
         provider1 = providers_fixture[0]
         provider2 = providers_fixture[1]
-
+        resource1 = resources_fixture[0]
+        resource2 = resources_fixture[1]
+        resource3 = resources_fixture[2]
         check_id = "cross_provider_latest_resources_total"
-        now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
 
-        FindingGroupDailySummary.objects.create(
+        latest_scan_provider1 = Scan.objects.create(
             tenant_id=provider1.tenant_id,
             provider=provider1,
-            check_id=check_id,
-            inserted_at=now - timedelta(days=1),
-            resources_total=20,
-            resources_fail=20,
-            fail_count=20,
+            state=StateChoices.COMPLETED,
+            trigger=Scan.TriggerChoices.MANUAL,
+            completed_at=datetime.now(timezone.utc),
         )
-        FindingGroupDailySummary.objects.create(
+
+        latest_scan_provider2 = Scan.objects.create(
             tenant_id=provider2.tenant_id,
             provider=provider2,
-            check_id=check_id,
-            inserted_at=now,
-            resources_total=7,
-            resources_fail=7,
-            fail_count=7,
+            state=StateChoices.COMPLETED,
+            trigger=Scan.TriggerChoices.MANUAL,
+            completed_at=datetime.now(timezone.utc),
         )
 
+        older_scan_provider1 = Scan.objects.create(
+            tenant_id=provider1.tenant_id,
+            provider=provider1,
+            state=StateChoices.COMPLETED,
+            trigger=Scan.TriggerChoices.MANUAL,
+            completed_at=datetime.now(timezone.utc) - timedelta(days=1),
+        )
+
+        # Older scan — these should be excluded from /latest
+        Finding.objects.create(
+            tenant_id=provider1.tenant_id,
+            uid="old_cross_provider_1",
+            scan=older_scan_provider1,
+            delta="new",
+            status="FAIL",
+            severity="high",
+            impact="high",
+            check_id=check_id,
+            check_metadata={"CheckId": check_id, "checktitle": "Cross provider check"},
+            first_seen_at=datetime.now(timezone.utc) - timedelta(days=2),
+            muted=False,
+        )
+
+        # Latest scan provider1: TWO findings (PASS + FAIL) for the same check
+        latest_p1_pass = Finding.objects.create(
+            tenant_id=provider1.tenant_id,
+            uid="latest_cross_provider_1_pass",
+            scan=latest_scan_provider1,
+            delta="new",
+            status="PASS",
+            severity="high",
+            impact="high",
+            check_id=check_id,
+            check_metadata={"CheckId": check_id, "checktitle": "Cross provider check"},
+            first_seen_at=datetime.now(timezone.utc) - timedelta(hours=1),
+            muted=False,
+        )
+        latest_p1_pass.add_resources([resource1])
+
+        latest_p1_fail = Finding.objects.create(
+            tenant_id=provider1.tenant_id,
+            uid="latest_cross_provider_1_fail",
+            scan=latest_scan_provider1,
+            delta="new",
+            status="FAIL",
+            severity="high",
+            impact="high",
+            check_id=check_id,
+            check_metadata={"CheckId": check_id, "checktitle": "Cross provider check"},
+            first_seen_at=datetime.now(timezone.utc) - timedelta(hours=1),
+            muted=False,
+        )
+        latest_p1_fail.add_resources([resource2])
+
+        # Latest scan provider2: one finding
+        latest_p2 = Finding.objects.create(
+            tenant_id=provider2.tenant_id,
+            uid="latest_cross_provider_2",
+            scan=latest_scan_provider2,
+            delta="new",
+            status="FAIL",
+            severity="high",
+            impact="high",
+            check_id=check_id,
+            check_metadata={"CheckId": check_id, "checktitle": "Cross provider check"},
+            first_seen_at=datetime.now(timezone.utc) - timedelta(hours=1),
+            muted=False,
+        )
+        latest_p2.add_resources([resource3])
+
         response = authenticated_client.get(
             reverse("finding-group-latest"),
-            {"filter[check_id]": check_id},
+            {"filter[check_id]": check_id, "filter[delta]": "new"},
         )
 
         assert response.status_code == status.HTTP_200_OK
         data = response.json()["data"]
         assert len(data) == 1
         attrs = data[0]["attributes"]
-        assert attrs["resources_total"] == 27
-        assert attrs["resources_fail"] == 27
-        assert attrs["fail_count"] == 27
+        # 3 findings total: 2 from provider1 latest + 1 from provider2 latest
+        assert attrs["pass_count"] == 1
+        assert attrs["fail_count"] == 2
+        assert attrs["resources_total"] == 3
+        assert attrs["resources_fail"] == 2
 
     def test_finding_groups_latest_provider_type_filter(
         self, authenticated_client, finding_groups_fixture
@@ -16136,6 +16355,31 @@ class TestFindingGroupViewSet:
         check_titles = [item["attributes"]["check_title"] for item in data]
         assert check_titles == sorted(check_titles)
 
+    @pytest.mark.parametrize(
+        "endpoint_name", ["finding-group-list", "finding-group-latest"]
+    )
+    @pytest.mark.parametrize(
+        "sort_field",
+        ["first_seen_at", "-first_seen_at", "last_seen_at", "failing_since"],
+    )
+    def test_finding_groups_sort_by_time_fields(
+        self,
+        authenticated_client,
+        finding_groups_fixture,
+        endpoint_name,
+        sort_field,
+    ):
+        """Test sorting by aggregated time fields (first_seen_at, last_seen_at, failing_since)."""
+        params = {"sort": sort_field}
+        if endpoint_name == "finding-group-list":
+            params["filter[inserted_at]"] = TODAY
+
+        response = authenticated_client.get(reverse(endpoint_name), params)
+
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) > 0
+
     def test_finding_groups_latest_ignores_date_filters(
         self, authenticated_client, finding_groups_fixture
     ):

api/src/backend/api/v1/serializers.py

@@ -4180,6 +4180,7 @@ class FindingGroupResourceSerializer(BaseSerializerV1):
     severity = serializers.CharField()
     first_seen_at = serializers.DateTimeField(required=False, allow_null=True)
     last_seen_at = serializers.DateTimeField(required=False, allow_null=True)
+    muted_reason = serializers.CharField(required=False, allow_null=True)
 
     class JSONAPIMeta:
         resource_name = "finding-group-resources"

api/src/backend/api/v1/views.py

@@ -48,7 +48,7 @@ from django.db.models import (
     When,
     Window,
 )
-from django.db.models.functions import Coalesce, RowNumber
+from django.db.models.functions import Cast, Coalesce, DenseRank, RowNumber
 from django.http import HttpResponse, QueryDict
 from django.shortcuts import redirect
 from django.urls import reverse
@@ -6788,13 +6788,13 @@ class FindingGroupViewSet(BaseRLSViewSet):
     security analysts to see which checks are failing across their
     infrastructure without scrolling through thousands of individual findings.
 
-    Uses pre-aggregated FindingGroupDailySummary table for efficient queries.
-    Daily summaries are re-aggregated across the requested date range.
+    Uses a hybrid strategy: pre-aggregated daily summaries when possible,
+    and raw findings when finding-level filters require precise subset metrics.
     """
 
     queryset = FindingGroupDailySummary.objects.all()
     serializer_class = FindingGroupSerializer
-    filterset_class = FindingGroupSummaryFilter
+    filterset_class = FindingGroupFilter
     filter_backends = [
         jsonapi_filters.QueryParameterValidationFilter,
         jsonapi_filters.OrderingFilter,
@@ -6813,12 +6813,12 @@ class FindingGroupViewSet(BaseRLSViewSet):
         affects the OpenAPI schema generated by drf-spectacular.
         """
         if self.action == "latest":
-            return LatestFindingGroupSummaryFilter
+            return LatestFindingGroupFilter
         if self.action == "resources":
             return FindingGroupFilter
         if self.action == "latest_resources":
             return LatestFindingGroupFilter
-        return FindingGroupSummaryFilter
+        return FindingGroupFilter
 
     def get_queryset(self):
         """Get the base FindingGroupDailySummary queryset with RLS filtering."""
@@ -6874,39 +6874,6 @@ class FindingGroupViewSet(BaseRLSViewSet):
         "resource_type__icontains": "type__icontains",
     }
 
-    def _get_finding_level_filter_keys(self, latest: bool = False) -> set[str]:
-        """Derive filters that require finding-level prefiltering."""
-        summary_filterset = (
-            LatestFindingGroupSummaryFilter if latest else FindingGroupSummaryFilter
-        )
-        finding_filterset = LatestFindingGroupFilter if latest else FindingGroupFilter
-
-        summary_supported = set(summary_filterset.base_filters.keys()) | {
-            "status",
-            "status__in",
-            "severity",
-            "severity__in",
-        }
-        finding_supported = set(finding_filterset.base_filters.keys())
-        return finding_supported - summary_supported
-
-    def _requires_finding_level_aggregation(
-        self, params: QueryDict, latest: bool = False
-    ) -> bool:
-        finding_level_keys = self._get_finding_level_filter_keys(latest=latest)
-        return any(key in finding_level_keys for key in params.keys())
-
-    def _get_matching_check_ids_from_findings(
-        self, normalized_params: QueryDict, latest: bool = False
-    ):
-        """Return check_ids matching advanced finding-level filters."""
-        finding_queryset = self._get_finding_queryset()
-        filterset_class = LatestFindingGroupFilter if latest else FindingGroupFilter
-        filterset = filterset_class(normalized_params, queryset=finding_queryset)
-        if not filterset.is_valid():
-            raise ValidationError(filterset.errors)
-        return filterset.qs.values("check_id").distinct()
-
     def _split_resource_filters(self, params: QueryDict) -> tuple[QueryDict, QueryDict]:
         resource_keys = set(self.RESOURCE_FILTER_MAP)
         finding_params = QueryDict(mutable=True)
@@ -6958,20 +6925,27 @@ class FindingGroupViewSet(BaseRLSViewSet):
 
         return filterset.qs.values("id")
 
+    def _get_finding_level_filter_keys(self, latest: bool = False) -> set[str]:
+        """Derive filters that require querying raw findings."""
+        summary_filterset = (
+            LatestFindingGroupSummaryFilter if latest else FindingGroupSummaryFilter
+        )
+        finding_filterset = LatestFindingGroupFilter if latest else FindingGroupFilter
+
+        summary_supported = set(summary_filterset.base_filters.keys())
+        finding_supported = set(finding_filterset.base_filters.keys())
+        return finding_supported - summary_supported
+
+    def _requires_finding_level_aggregation(
+        self, params: QueryDict, latest: bool = False
+    ) -> bool:
+        finding_level_keys = self._get_finding_level_filter_keys(latest=latest)
+        return any(key in finding_level_keys for key in params.keys())
+
     def _aggregate_daily_summaries(self, queryset):
-        """
-        Re-aggregate daily summaries across the date range.
-
-        Takes pre-computed daily summaries and aggregates them by check_id
-        to produce totals across the selected date range.
-        """
-        from django.db.models import CharField
-        from django.db.models.functions import Cast
-
+        """Re-aggregate summary rows by check_id."""
         return queryset.values("check_id").annotate(
-            # Max severity across days
             severity_order=Max("severity_order"),
-            # Sum counts across days
             pass_count=Sum("pass_count"),
             fail_count=Sum("fail_count"),
             muted_count=Sum("muted_count"),
@@ -6979,22 +6953,99 @@ class FindingGroupViewSet(BaseRLSViewSet):
             changed_count=Sum("changed_count"),
             resources_total=Sum("resources_total"),
             resources_fail=Sum("resources_fail"),
-            # Collect provider types using StringAgg (cast enum to text first)
             impacted_providers_str=StringAgg(
                 Cast("provider__provider", CharField()),
                 delimiter=",",
                 distinct=True,
                 default="",
             ),
-            # Min/Max timing across days
-            first_seen_at=Min("first_seen_at"),
-            last_seen_at=Max("last_seen_at"),
-            failing_since=Min("failing_since"),
-            # Get check metadata from first row (same for all days)
+            agg_first_seen_at=Min("first_seen_at"),
+            agg_last_seen_at=Max("last_seen_at"),
+            agg_failing_since=Min("failing_since"),
             check_title=Max("check_title"),
             check_description=Max("check_description"),
         )
 
+    def _aggregate_findings(self, queryset):
+        """Aggregate findings by check_id for finding-group endpoints."""
+        severity_case = Case(
+            *[
+                When(severity=severity, then=Value(order))
+                for severity, order in SEVERITY_ORDER.items()
+            ],
+            output_field=IntegerField(),
+        )
+
+        return queryset.values("check_id").annotate(
+            severity_order=Max(severity_case),
+            pass_count=Count("id", filter=Q(status="PASS", muted=False)),
+            fail_count=Count("id", filter=Q(status="FAIL", muted=False)),
+            muted_count=Count("id", filter=Q(muted=True)),
+            new_count=Count("id", filter=Q(delta="new", muted=False)),
+            changed_count=Count("id", filter=Q(delta="changed", muted=False)),
+            resources_total=Count("resources__id", distinct=True),
+            resources_fail=Count(
+                "resources__id",
+                distinct=True,
+                filter=Q(status="FAIL", muted=False),
+            ),
+            impacted_providers_str=StringAgg(
+                Cast("scan__provider__provider", CharField()),
+                delimiter=",",
+                distinct=True,
+                default="",
+            ),
+            agg_first_seen_at=Min("first_seen_at"),
+            agg_last_seen_at=Max("inserted_at"),
+            agg_failing_since=Min(
+                "first_seen_at", filter=Q(status="FAIL", muted=False)
+            ),
+            check_title=Coalesce(
+                Max(Cast("check_metadata__CheckTitle", CharField())),
+                Max(Cast("check_metadata__checktitle", CharField())),
+                Max(Cast("check_metadata__Checktitle", CharField())),
+            ),
+            check_description=Coalesce(
+                Max(Cast("check_metadata__Description", CharField())),
+                Max(Cast("check_metadata__description", CharField())),
+            ),
+        )
+
+    def _split_computed_aggregate_filters(
+        self, params: QueryDict
+    ) -> tuple[QueryDict, QueryDict]:
+        """Split finding filters from computed aggregate filters."""
+        computed_keys = {"status", "status__in", "severity", "severity__in"}
+        finding_params = QueryDict(mutable=True)
+        computed_params = QueryDict(mutable=True)
+
+        for key, values in params.lists():
+            if key in computed_keys:
+                computed_params.setlist(key, values)
+            else:
+                finding_params.setlist(key, values)
+
+        return finding_params, computed_params
+
+    def _get_latest_findings_per_check_provider(self, filtered_queryset):
+        """Keep all findings from the latest scan per (check_id, provider)."""
+        latest_ids = (
+            filtered_queryset.annotate(
+                scan_rank=Window(
+                    expression=DenseRank(),
+                    partition_by=[F("check_id"), F("scan__provider_id")],
+                    order_by=[
+                        F("scan__completed_at").desc(nulls_last=True),
+                        F("scan_id").desc(),
+                    ],
+                )
+            )
+            .filter(scan_rank=1)
+            .values("id")
+        )
+
+        return filtered_queryset.filter(id__in=Subquery(latest_ids))
+
     def _post_process_aggregation(self, aggregated_data):
         """
         Post-process aggregation results to add computed fields.
@@ -7011,6 +7062,13 @@ class FindingGroupViewSet(BaseRLSViewSet):
                 severity_order, "informational"
             )
 
+            if "agg_first_seen_at" in row:
+                row["first_seen_at"] = row.pop("agg_first_seen_at")
+            if "agg_last_seen_at" in row:
+                row["last_seen_at"] = row.pop("agg_last_seen_at")
+            if "agg_failing_since" in row:
+                row["failing_since"] = row.pop("agg_failing_since")
+
             # Compute aggregated status
             if row.get("fail_count", 0) > 0:
                 row["status"] = "FAIL"
@@ -7042,9 +7100,9 @@ class FindingGroupViewSet(BaseRLSViewSet):
             "changed_count": "changed_count",
             "resources_total": "resources_total",
             "resources_fail": "resources_fail",
-            "first_seen_at": "first_seen_at",
-            "last_seen_at": "last_seen_at",
-            "failing_since": "failing_since",
+            "first_seen_at": "agg_first_seen_at",
+            "last_seen_at": "agg_last_seen_at",
+            "failing_since": "agg_failing_since",
         }
 
         ordering = []
@@ -7071,26 +7129,12 @@ class FindingGroupViewSet(BaseRLSViewSet):
 
         return ordering
 
-    def _apply_aggregated_computed_filters(
-        self, queryset, normalized_params: QueryDict
-    ):
+    def _apply_aggregated_computed_filters(self, queryset, computed_params: QueryDict):
         """Apply computed filters (status/severity) on aggregated finding-group rows."""
-        status_params = QueryDict(mutable=True)
-        if normalized_params.get("status"):
-            status_params.setlist("status", [normalized_params.get("status")])
-        if normalized_params.getlist("status__in"):
-            status_params.setlist("status__in", normalized_params.getlist("status__in"))
-        if normalized_params.get("severity"):
-            status_params.setlist("severity", [normalized_params.get("severity")])
-        if normalized_params.getlist("severity__in"):
-            status_params.setlist(
-                "severity__in", normalized_params.getlist("severity__in")
-            )
-
-        if not status_params:
+        if not computed_params:
             return queryset
 
-        if status_params.get("status") or status_params.getlist("status__in"):
+        if computed_params.get("status") or computed_params.getlist("status__in"):
             queryset = queryset.annotate(
                 aggregated_status=Case(
                     When(fail_count__gt=0, then=Value("FAIL")),
@@ -7101,7 +7145,7 @@ class FindingGroupViewSet(BaseRLSViewSet):
             )
 
         filterset = FindingGroupAggregatedComputedFilter(
-            status_params, queryset=queryset
+            computed_params, queryset=queryset
         )
         if not filterset.is_valid():
             raise ValidationError(filterset.errors)
@@ -7180,6 +7224,7 @@ class FindingGroupViewSet(BaseRLSViewSet):
                 ),
                 first_seen_at=Min("finding__first_seen_at"),
                 last_seen_at=Max("finding__inserted_at"),
+                muted_reason=Max("finding__muted_reason"),
             )
             .filter(resource_id__isnull=False)
             .order_by("resource_id")
@@ -7215,67 +7260,90 @@ class FindingGroupViewSet(BaseRLSViewSet):
                     ),
                     "first_seen_at": row["first_seen_at"],
                     "last_seen_at": row["last_seen_at"],
+                    "muted_reason": row.get("muted_reason"),
                 }
             )
 
         return results
 
+    def _build_aggregated_queryset(self, finding_params, latest=False):
+        """Select the summary or findings path and return an aggregated queryset."""
+        finding_filterset_class = (
+            LatestFindingGroupFilter if latest else FindingGroupFilter
+        )
+        summary_filterset_class = (
+            LatestFindingGroupSummaryFilter if latest else FindingGroupSummaryFilter
+        )
+
+        if self._requires_finding_level_aggregation(finding_params, latest=latest):
+            finding_queryset = self._get_finding_queryset()
+            filterset = finding_filterset_class(
+                finding_params, queryset=finding_queryset
+            )
+            if not filterset.is_valid():
+                raise ValidationError(filterset.errors)
+            filtered_queryset = filterset.qs
+            if latest:
+                filtered_queryset = self._get_latest_findings_per_check_provider(
+                    filtered_queryset
+                )
+            return self._aggregate_findings(filtered_queryset)
+
+        summary_queryset = self.get_queryset()
+        filterset = summary_filterset_class(finding_params, queryset=summary_queryset)
+        if not filterset.is_valid():
+            raise ValidationError(filterset.errors)
+        filtered_queryset = filterset.qs
+        if latest:
+            latest_per_check_ids = (
+                filtered_queryset.order_by("check_id", "provider_id", "-inserted_at")
+                .distinct("check_id", "provider_id")
+                .values("id")
+            )
+            filtered_queryset = filtered_queryset.filter(
+                id__in=Subquery(latest_per_check_ids)
+            )
+        return self._aggregate_daily_summaries(filtered_queryset)
+
+    def _sorted_paginated_response(self, request, aggregated_queryset):
+        """Apply ordering, pagination, post-processing, and return the Response."""
+        sort_param = request.query_params.get("sort")
+        if sort_param:
+            ordering = self._validate_sort_fields(sort_param)
+            if ordering:
+                aggregated_queryset = aggregated_queryset.order_by(*ordering)
+        else:
+            aggregated_queryset = aggregated_queryset.order_by(
+                "-fail_count", "-severity_order", "check_id"
+            )
+
+        page = self.paginate_queryset(aggregated_queryset)
+        if page is not None:
+            processed_data = self._post_process_aggregation(page)
+            serializer = self.get_serializer(processed_data, many=True)
+            return self.get_paginated_response(serializer.data)
+
+        processed_data = self._post_process_aggregation(aggregated_queryset)
+        serializer = self.get_serializer(processed_data, many=True)
+        return Response(serializer.data)
+
     def list(self, request, *args, **kwargs):
         """
         List finding groups with aggregation and filtering.
 
         Returns findings grouped by check_id with aggregated metrics.
         Requires at least one date filter for performance.
-        Uses pre-aggregated daily summaries for efficient queries.
+        Uses summaries when possible and raw findings for finding-level filters.
         """
-        queryset = self.get_queryset()
-
-        # Apply filters
         normalized_params = self._normalize_jsonapi_params(request.query_params)
-        filterset = self.filterset_class(normalized_params, queryset=queryset)
-        if not filterset.is_valid():
-            raise ValidationError(filterset.errors)
-        filtered_queryset = filterset.qs
-
-        # Re-aggregate daily summaries across the date range
-        aggregated_queryset = self._aggregate_daily_summaries(filtered_queryset)
-        aggregated_queryset = self._apply_aggregated_computed_filters(
-            aggregated_queryset, normalized_params
+        finding_params, computed_params = self._split_computed_aggregate_filters(
+            normalized_params
         )
-
-        if self._requires_finding_level_aggregation(normalized_params, latest=False):
-            matching_check_ids = self._get_matching_check_ids_from_findings(
-                normalized_params, latest=False
-            )
-            aggregated_queryset = aggregated_queryset.filter(
-                check_id__in=Subquery(matching_check_ids)
-            )
-
-        # Apply ordering (respect JSON:API sort param or use default)
-        sort_param = request.query_params.get("sort")
-        if sort_param:
-            # Convert JSON:API sort notation (prefix '-' for descending)
-            ordering = self._validate_sort_fields(sort_param)
-            if ordering:
-                aggregated_queryset = aggregated_queryset.order_by(*ordering)
-        else:
-            # Default ordering: failures first, then severity, then check_id
-            aggregated_queryset = aggregated_queryset.order_by(
-                "-fail_count", "-severity_order", "check_id"
-            )
-
-        # Paginate
-        page = self.paginate_queryset(aggregated_queryset)
-        if page is not None:
-            # Post-process the page
-            processed_data = self._post_process_aggregation(page)
-            serializer = self.get_serializer(processed_data, many=True)
-            return self.get_paginated_response(serializer.data)
-
-        # Post-process all results (no pagination)
-        processed_data = self._post_process_aggregation(aggregated_queryset)
-        serializer = self.get_serializer(processed_data, many=True)
-        return Response(serializer.data)
+        aggregated_qs = self._build_aggregated_queryset(finding_params, latest=False)
+        aggregated_qs = self._apply_aggregated_computed_filters(
+            aggregated_qs, computed_params
+        )
+        return self._sorted_paginated_response(request, aggregated_qs)
 
     @extend_schema(
         summary="List latest finding groups",
@@ -7293,69 +7361,22 @@ class FindingGroupViewSet(BaseRLSViewSet):
         """
         List the latest finding group state per check_id.
 
-        Returns findings grouped by check_id using the latest available
-        inserted_at date per check_id, without requiring date filters.
+        Returns findings grouped by check_id using latest data per
+        (check_id, provider), without requiring date filters.
         """
-        queryset = self.get_queryset()
-
-        # Apply other filters (provider_id, provider_type, check_id, etc.)
         normalized_params = self._normalize_jsonapi_params(request.query_params)
-        # Remove date filters since we're using latest
         for key in list(normalized_params.keys()):
             if key.startswith("inserted_at"):
                 del normalized_params[key]
 
-        filterset_class = self.get_filterset_class()
-        filterset = filterset_class(normalized_params, queryset=queryset)
-        if not filterset.is_valid():
-            raise ValidationError(filterset.errors)
-        filtered_queryset = filterset.qs
-
-        # Keep only the latest row per (check_id, provider), then aggregate by check_id.
-        latest_per_check_ids = (
-            filtered_queryset.order_by("check_id", "provider_id", "-inserted_at")
-            .distinct("check_id", "provider_id")
-            .values("id")
+        finding_params, computed_params = self._split_computed_aggregate_filters(
+            normalized_params
         )
-        latest_per_check = filtered_queryset.filter(
-            id__in=Subquery(latest_per_check_ids)
+        aggregated_qs = self._build_aggregated_queryset(finding_params, latest=True)
+        aggregated_qs = self._apply_aggregated_computed_filters(
+            aggregated_qs, computed_params
         )
-
-        # Re-aggregate daily summaries
-        aggregated_queryset = self._aggregate_daily_summaries(latest_per_check)
-        aggregated_queryset = self._apply_aggregated_computed_filters(
-            aggregated_queryset, normalized_params
-        )
-
-        if self._requires_finding_level_aggregation(normalized_params, latest=True):
-            matching_check_ids = self._get_matching_check_ids_from_findings(
-                normalized_params, latest=True
-            )
-            aggregated_queryset = aggregated_queryset.filter(
-                check_id__in=Subquery(matching_check_ids)
-            )
-
-        # Apply ordering
-        sort_param = request.query_params.get("sort")
-        if sort_param:
-            ordering = self._validate_sort_fields(sort_param)
-            if ordering:
-                aggregated_queryset = aggregated_queryset.order_by(*ordering)
-        else:
-            aggregated_queryset = aggregated_queryset.order_by(
-                "-fail_count", "-severity_order", "check_id"
-            )
-
-        # Paginate
-        page = self.paginate_queryset(aggregated_queryset)
-        if page is not None:
-            processed_data = self._post_process_aggregation(page)
-            serializer = self.get_serializer(processed_data, many=True)
-            return self.get_paginated_response(serializer.data)
-
-        processed_data = self._post_process_aggregation(aggregated_queryset)
-        serializer = self.get_serializer(processed_data, many=True)
-        return Response(serializer.data)
+        return self._sorted_paginated_response(request, aggregated_qs)
 
     @extend_schema(
         summary="List resources for a finding group",