diff --git a/lib/ash/actions/aggregate.ex b/lib/ash/actions/aggregate.ex
index d0f647b6..4643df02 100644
--- a/lib/ash/actions/aggregate.ex
+++ b/lib/ash/actions/aggregate.ex
@@ -8,84 +8,90 @@ defmodule Ash.Actions.Aggregate do
     action = query.action || Ash.Resource.Info.primary_action!(query.resource, :read)
     opts = Keyword.put_new(opts, :read_action, action.name)
 
-    with {:ok, aggregates} <- validate_aggregates(query, aggregates, opts),
-         %{valid?: true} = query <- Ash.Actions.Read.handle_attribute_multitenancy(query) do
+    with %{valid?: true} = query <- Ash.Actions.Read.handle_attribute_multitenancy(query) do
       aggregates
-      |> Enum.group_by(fn aggregate ->
-        agg_authorize? = aggregate.authorize? && opts[:authorize?]
+      |> Enum.group_by(fn
+        %Ash.Query.Aggregate{} = aggregate ->
+          agg_authorize? = aggregate.authorize? && opts[:authorize?]
 
-        read_action =
-          aggregate.read_action || query.action ||
-            Ash.Resource.Info.primary_action!(query.resource, :read).name
+          read_action =
+            aggregate.read_action || query.action ||
+              Ash.Resource.Info.primary_action!(query.resource, :read).name
 
-        {agg_authorize?, read_action}
+          {agg_authorize?, read_action}
+
+        {_name, _kind} ->
+          {!!opts[:authorize?], opts[:read_action]}
+
+        {_name, _kind, agg_opts} ->
+          authorize? =
+            Keyword.get(agg_opts, :authorize?, true) && opts[:authorize?]
+
+          {authorize?, agg_opts[:read_action] || opts[:read_action] || action.name}
       end)
-      |> Enum.reduce_while({:ok, %{}}, fn {{agg_authorize?, read_action}, aggregates},
-                                          {:ok, acc} ->
-        query =
-          if query.__validated_for_action__ == read_action do
-            query
-          else
-            Ash.Query.for_read(query, read_action, %{},
-              tenant: opts[:tenant],
-              actor: opts[:actor],
-              authorize?: opts[:authorize?]
-            )
-          end
-
-        query = %{query | api: api}
-
-        Ash.Tracer.span :action,
-                        Ash.Api.Info.span_name(query.api, query.resource, :aggregate),
-                        opts[:tracer] do
-          metadata = %{
-            api: query.api,
-            resource: query.resource,
-            resource_short_name: Ash.Resource.Info.short_name(query.resource),
-            aggregates: List.wrap(aggregates),
-            actor: opts[:actor],
-            tenant: opts[:tenant],
-            action: action.name,
-            authorize?: opts[:authorize?]
-          }
-
-          Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(query.api), :aggregate],
-                                    metadata do
-            Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
-            query = Map.put(query, :aggregates, Map.new(aggregates, &{&1.name, &1}))
-
-            with {:ok, query} <- authorize_query(query, opts, agg_authorize?),
-                 {:ok, data_layer_query} <-
-                   Ash.Query.data_layer_query(Ash.Query.new(query.resource)),
-                 aggregates <- merge_query_into_aggregates(query, aggregates),
-                 {:ok, result} <-
-                   Ash.DataLayer.run_aggregate_query(data_layer_query, aggregates, query.resource) do
-              {:cont, {:ok, Map.merge(acc, result)}}
+      |> Enum.reduce_while({:ok, %{}}, fn
+        {{agg_authorize?, read_action}, aggregates}, {:ok, acc} ->
+          query =
+            if query.__validated_for_action__ == read_action do
+              query
             else
-              {:error, error} ->
-                {:halt, {:error, error}}
+              Ash.Query.for_read(query, read_action, %{},
+                tenant: opts[:tenant],
+                actor: opts[:actor],
+                authorize?: opts[:authorize?]
+              )
+            end
+
+          query = %{query | api: api}
+
+          Ash.Tracer.span :action,
+                          Ash.Api.Info.span_name(query.api, query.resource, :aggregate),
+                          opts[:tracer] do
+            metadata = %{
+              api: query.api,
+              resource: query.resource,
+              resource_short_name: Ash.Resource.Info.short_name(query.resource),
+              aggregates: List.wrap(aggregates),
+              actor: opts[:actor],
+              tenant: opts[:tenant],
+              action: action.name,
+              authorize?: opts[:authorize?]
+            }
+
+            Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(query.api), :aggregate],
+                                      metadata do
+              Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
+
+              with {:ok, query} <- authorize_query(query, opts, agg_authorize?),
+                   {:ok, aggregates} <- validate_aggregates(query, aggregates, opts),
+                   {:ok, data_layer_query} <-
+                     Ash.Query.data_layer_query(Ash.Query.new(query.resource)),
+                   {:ok, result} <-
+                     Ash.DataLayer.run_aggregate_query(
+                       data_layer_query,
+                       aggregates,
+                       query.resource
+                     ) do
+                {:cont, {:ok, Map.merge(acc, result)}}
+              else
+                {:error, error} ->
+                  {:halt, {:error, error}}
+              end
             end
           end
-        end
       end)
     end
   end
 
-  defp merge_query_into_aggregates(query, aggregates) do
-    Enum.map(aggregates, fn aggregate ->
-      %{
-        aggregate
-        | query:
-            aggregate.query
-            |> Ash.Query.do_filter(query.filter)
-            |> Ash.Query.sort(query.sort, prepend?: true)
-            |> Ash.Query.distinct_sort(query.distinct_sort, prepend?: true)
-            |> Ash.Query.limit(query.limit)
-            |> Ash.Query.set_tenant(query.tenant)
-            |> merge_offset(query.offset)
-            |> Ash.Query.set_context(query.context)
-      }
-    end)
+  defp merge_query(left, right) do
+    left
+    |> Ash.Query.do_filter(right.filter)
+    |> Ash.Query.sort(right.sort, prepend?: true)
+    |> Ash.Query.distinct_sort(right.distinct_sort, prepend?: true)
+    |> Ash.Query.limit(right.limit)
+    |> Ash.Query.set_tenant(right.tenant)
+    |> merge_offset(right.offset)
+    |> Ash.Query.set_context(right.context)
   end
 
   defp merge_offset(query, offset) do
@@ -128,7 +134,7 @@ defmodule Ash.Actions.Aggregate do
                query.resource,
                name,
                kind,
-               set_opts([], opts)
+               set_opts(query, [], opts)
              ) do
           {:ok, aggregate} ->
             {:cont, {:ok, [aggregate | aggregates]}}
@@ -138,7 +144,7 @@ defmodule Ash.Actions.Aggregate do
         end
 
       {name, kind, agg_opts}, {:ok, aggregates} ->
-        case Ash.Query.Aggregate.new(query.resource, name, kind, set_opts(agg_opts, opts)) do
+        case Ash.Query.Aggregate.new(query.resource, name, kind, set_opts(query, agg_opts, opts)) do
           {:ok, aggregate} ->
             {:cont, {:ok, [aggregate | aggregates]}}
 
@@ -148,8 +154,23 @@ defmodule Ash.Actions.Aggregate do
     end)
   end
 
-  defp set_opts(specified, others) do
+  defp set_opts(query, specified, others) do
     {agg_opts, _} = Ash.Query.Aggregate.split_aggregate_opts(others)
-    Keyword.merge(agg_opts, specified)
+
+    query =
+      case agg_opts[:query] do
+        %Ash.Query{} = agg_query ->
+          merge_query(agg_query, query)
+
+        nil ->
+          query
+
+        opts ->
+          Ash.Query.build(query, opts)
+      end
+
+    agg_opts
+    |> Keyword.merge(specified)
+    |> Keyword.put(:query, query)
   end
 end
diff --git a/lib/ash/api/api.ex b/lib/ash/api/api.ex
index 82d6d6f1..4d37d06c 100644
--- a/lib/ash/api/api.ex
+++ b/lib/ash/api/api.ex
@@ -1623,16 +1623,36 @@ defmodule Ash.Api do
     end
   end
 
+  @doc """
+  Runs an aggregate or aggregates over a resource query
+
+  If you pass an `%Ash.Query.Aggregate{}`, gotten from `Ash.Query.Aggregate.new()`,
+  the query provided as the first argument to this function will not apply. For this
+  reason, it is preferred that you pass in the tuple format, i.e
+
+  Prefer this:
+  `Api.aggregate(query, {:count_of_things, :count})`
+
+  Over this:
+  `Api.aggregate(query, Ash.Query.Aggregate.new(...))`
+
+  #{Spark.OptionsHelpers.docs(@aggregate_opts)}
+  """
   @callback aggregate(
               Ash.Query.t(),
-              Ash.Api.aggregate() | list(Ash.Api.aggregate()),
+              aggregate() | list(aggregate()),
               opts :: Keyword.t()
             ) ::
               {:ok, any} | {:error, Ash.Error.t()}
 
+  @doc """
+  Runs an aggregate or aggregates over a resource query
+
+  See `c:aggregate/3` for more.
+  """
   @callback aggregate!(
               Ash.Query.t(),
-              Ash.Api.aggregate() | list(Ash.Api.aggregate()),
+              aggregate() | list(aggregate()),
               opts :: Keyword.t()
             ) ::
               any | no_return
diff --git a/lib/ash/api/interface.ex b/lib/ash/api/interface.ex
index 0b5463d9..a46d5fe7 100644
--- a/lib/ash/api/interface.ex
+++ b/lib/ash/api/interface.ex
@@ -81,15 +81,9 @@ defmodule Ash.Api.Interface do
 
         {aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
 
-        case Ash.Query.Aggregate.new(query.resource, :count, :count, aggregate_opts) do
-          {:ok, aggregate} ->
-            case Api.aggregate(__MODULE__, query, aggregate, opts) do
-              {:ok, %{count: count}} ->
-                count
-
-              {:error, error} ->
-                raise Ash.Error.to_error_class(error)
-            end
+        case Api.aggregate(__MODULE__, query, {:count, :count, aggregate_opts}, opts) do
+          {:ok, %{count: count}} ->
+            count
 
           {:error, error} ->
             raise Ash.Error.to_error_class(error)
@@ -108,15 +102,9 @@ defmodule Ash.Api.Interface do
 
         {aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
 
-        case Ash.Query.Aggregate.new(query.resource, :count, :count, aggregate_opts) do
-          {:ok, aggregate} ->
-            case Api.aggregate(__MODULE__, query, aggregate, opts) do
-              {:ok, %{count: count}} ->
-                {:ok, count}
-
-              {:error, error} ->
-                {:error, Ash.Error.to_error_class(error)}
-            end
+        case Api.aggregate(__MODULE__, query, {:count, :count, aggregate_opts}, opts) do
+          {:ok, %{count: count}} ->
+            {:ok, count}
 
           {:error, error} ->
             {:error, Ash.Error.to_error_class(error)}
@@ -135,15 +123,9 @@ defmodule Ash.Api.Interface do
 
         {aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
 
-        case Ash.Query.Aggregate.new(query.resource, :exists, :exists, aggregate_opts) do
-          {:ok, aggregate} ->
-            case Api.aggregate(__MODULE__, query, aggregate, opts) do
-              {:ok, %{exists: exists}} ->
-                exists
-
-              {:error, error} ->
-                raise Ash.Error.to_error_class(error)
-            end
+        case Api.aggregate(__MODULE__, query, {:exists, :exists, aggregate_opts}, opts) do
+          {:ok, %{exists: exists}} ->
+            exists
 
           {:error, error} ->
             raise Ash.Error.to_error_class(error)
@@ -162,15 +144,9 @@ defmodule Ash.Api.Interface do
 
         {aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
 
-        case Ash.Query.Aggregate.new(query.resource, :exists, :exists, aggregate_opts) do
-          {:ok, aggregate} ->
-            case Api.aggregate(__MODULE__, query, aggregate, opts) do
-              {:ok, %{exists: exists}} ->
-                {:ok, exists}
-
-              {:error, error} ->
-                {:error, Ash.Error.to_error_class(error)}
-            end
+        case Api.aggregate(__MODULE__, query, {:exists, :exists, aggregate_opts}, opts) do
+          {:ok, %{exists: exists}} ->
+            {:ok, exists}
 
           {:error, error} ->
             {:error, Ash.Error.to_error_class(error)}
@@ -190,20 +166,14 @@ defmodule Ash.Api.Interface do
 
           {aggregate_opts, opts} = Ash.Query.Aggregate.split_aggregate_opts(opts)
 
-          case Ash.Query.Aggregate.new(
-                 query.resource,
-                 unquote(kind),
-                 unquote(kind),
-                 Keyword.put(aggregate_opts, :field, field)
+          case Api.aggregate(
+                 __MODULE__,
+                 query,
+                 {unquote(kind), unquote(kind), Keyword.put(aggregate_opts, :field, field)},
+                 opts
                ) do
-            {:ok, aggregate} ->
-              case Api.aggregate(__MODULE__, query, aggregate, opts) do
-                {:ok, %{unquote(kind) => value}} ->
-                  {:ok, value}
-
-                {:error, error} ->
-                  {:error, Ash.Error.to_error_class(error)}
-              end
+            {:ok, %{unquote(kind) => value}} ->
+              {:ok, value}
 
             {:error, error} ->
               {:error, Ash.Error.to_error_class(error)}
@@ -223,20 +193,14 @@ defmodule Ash.Api.Interface do
               opts
             end
 
-          case Ash.Query.Aggregate.new(
-                 query.resource,
-                 unquote(kind),
-                 unquote(kind),
-                 Keyword.put(aggregate_opts, :field, field)
+          case Api.aggregate(
+                 __MODULE__,
+                 query,
+                 {unquote(kind), unquote(kind), Keyword.put(aggregate_opts, :field, field)},
+                 opts
                ) do
-            {:ok, aggregate} ->
-              case Api.aggregate(__MODULE__, query, aggregate, opts) do
-                {:ok, %{unquote(kind) => value}} ->
-                  value
-
-                {:error, error} ->
-                  raise Ash.Error.to_error_class(error)
-              end
+            {:ok, %{unquote(kind) => value}} ->
+              value
 
             {:error, error} ->
               raise Ash.Error.to_error_class(error)