fix: more cleanup

pkg/modules/llmpricingrule/impllmpricingrule/module.go

@@ -213,18 +213,18 @@ func (module *module) discoverModels(ctx context.Context, orgID valuer.UUID) ([]
 					Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
 						Name:   "A",
 						Signal: telemetrytypes.SignalTraces,
-						Filter: &qbtypes.Filter{Expression: fmt.Sprintf("%s EXISTS", llmpricingruletypes.GenAIRequestModel)},
+						Filter: &qbtypes.Filter{Expression: fmt.Sprintf("%s EXISTS", telemetrytypes.GenAIRequestModel)},
 						Aggregations: []qbtypes.TraceAggregation{
 							{Expression: "count()", Alias: "spanCount"},
 						},
 						GroupBy: []qbtypes.GroupByKey{
 							{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
-								Name:          llmpricingruletypes.GenAIRequestModel,
+								Name:          telemetrytypes.GenAIRequestModel,
 								FieldContext:  telemetrytypes.FieldContextSpan,
 								FieldDataType: telemetrytypes.FieldDataTypeString,
 							}},
 							{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
-								Name:          llmpricingruletypes.GenAIProviderName,
+								Name:          telemetrytypes.GenAIProviderName,
 								FieldContext:  telemetrytypes.FieldContextSpan,
 								FieldDataType: telemetrytypes.FieldDataTypeString,
 							}},
@@ -254,9 +254,9 @@ func (module *module) discoverModels(ctx context.Context, orgID valuer.UUID) ([]
 		switch c.Type {
 		case qbtypes.ColumnTypeGroup:
 			switch c.Name {
-			case llmpricingruletypes.GenAIRequestModel:
+			case telemetrytypes.GenAIRequestModel:
 				modelIdx = i
-			case llmpricingruletypes.GenAIProviderName:
+			case telemetrytypes.GenAIProviderName:
 				providerIdx = i
 			}
 		case qbtypes.ColumnTypeAggregation:

pkg/querier/querier.go

@@ -42,6 +42,7 @@ type querier struct {
 	metadataStore            telemetrytypes.MetadataStore
 	promEngine               prometheus.Prometheus
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
+	aiTraceStmtBuilder       qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
@@ -61,6 +62,7 @@ func New(
 	metadataStore telemetrytypes.MetadataStore,
 	promEngine prometheus.Prometheus,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
+	aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
@@ -82,6 +84,7 @@ func New(
 		metadataStore:            metadataStore,
 		promEngine:               promEngine,
 		traceStmtBuilder:         traceStmtBuilder,
+		aiTraceStmtBuilder:       aiTraceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
 		auditStmtBuilder:         auditStmtBuilder,
 		metricStmtBuilder:        metricStmtBuilder,
@@ -235,10 +238,18 @@ func (q *querier) buildQueries(
 			case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-				bq := newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
+				stmtBuilder := q.traceStmtBuilder
+				if spec.Source == telemetrytypes.SourceAI {
+					event.Source = telemetrytypes.SourceAI.StringValue()
+					stmtBuilder = q.aiTraceStmtBuilder
+				}
+				bq := newBuilderQuery(q.logger, q.telemetryStore, stmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
+				if spec.Source == telemetrytypes.SourceAI {
+					return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not logs")
+				}
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
 				stmtBuilder := q.logStmtBuilder
@@ -249,6 +260,9 @@ func (q *querier) buildQueries(
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
+				if spec.Source == telemetrytypes.SourceAI {
+					return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "source \"ai\" is only supported for the traces signal, not metrics")
+				}
 				// Spec was already patched by resolveMetricMetadata. Queries
 				// whose every aggregation was missing live in
 				// missingMetricQuerySet and produce empty preseeded results
@@ -858,7 +872,11 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
 		specCopy := qt.spec.Copy()
 		specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
 		adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
-		return newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
+		shiftStmtBuilder := q.traceStmtBuilder
+		if qt.spec.Source == telemetrytypes.SourceAI {
+			shiftStmtBuilder = q.aiTraceStmtBuilder
+		}
+		return newBuilderQuery(q.logger, q.telemetryStore, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
 
 	case *builderQuery[qbtypes.LogAggregation]:
 		specCopy := qt.spec.Copy()

pkg/querier/querier_test.go

@@ -49,6 +49,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		metadataStore,
 		nil,                // prometheus
 		nil,                // traceStmtBuilder
+		nil,                // aiTraceStmtBuilder
 		nil,                // logStmtBuilder
 		nil,                // auditStmtBuilder
 		nil,                // metricStmtBuilder
@@ -121,6 +122,7 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		metadataStore,
 		nil,                      // prometheus
 		nil,                      // traceStmtBuilder
+		nil,                      // aiTraceStmtBuilder
 		nil,                      // logStmtBuilder
 		nil,                      // auditStmtBuilder
 		&mockMetricStmtBuilder{}, // metricStmtBuilder

pkg/querier/signozquerier/provider.go

@@ -9,6 +9,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryai"
 	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetadata"
@@ -92,6 +93,30 @@ func newProvider(
 		cfg.SkipResourceFingerprint.Threshold,
 	)
 
+	// AI trace statement builder (source=ai). The gen_ai gate/column keys are
+	// surfaced by the metadata store itself (enrichWithGenAIKeys), so queries work
+	// before any gen_ai metadata is ingested — no per-builder decoration needed.
+	aiBaseCondition := telemetryai.NewGenAIBaseConditionProvider()
+	aiDelegateTraceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
+		settings,
+		telemetryMetadataStore,
+		traceFieldMapper,
+		traceConditionBuilder,
+		traceAggExprRewriter,
+		telemetryStore,
+		flagger,
+		cfg.SkipResourceFingerprint.Enabled,
+		cfg.SkipResourceFingerprint.Threshold,
+	)
+	aiTraceStmtBuilder := telemetryai.NewAITraceStatementBuilder(
+		settings,
+		telemetryMetadataStore,
+		traceFieldMapper,
+		traceConditionBuilder,
+		aiBaseCondition,
+		aiDelegateTraceStmtBuilder,
+	)
+
 	// Create trace operator statement builder
 	traceOperatorStmtBuilder := telemetrytraces.NewTraceOperatorStatementBuilder(
 		settings,
@@ -185,6 +210,7 @@ func newProvider(
 		telemetryMetadataStore,
 		prometheus,
 		traceStmtBuilder,
+		aiTraceStmtBuilder,
 		logStmtBuilder,
 		auditStmtBuilder,
 		metricStmtBuilder,

pkg/query-service/rules/setups_test.go

@@ -48,6 +48,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		metadataStore,
 		nil, // prometheus
 		nil, // traceStmtBuilder
+		nil, // aiTraceStmtBuilder
 		nil, // logStmtBuilder
 		nil, // auditStmtBuilder
 		metricStmtBuilder,
@@ -103,6 +104,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		metadataStore,
 		nil,            // prometheus
 		nil,            // traceStmtBuilder
+		nil,            // aiTraceStmtBuilder
 		logStmtBuilder, // logStmtBuilder
 		nil,            // auditStmtBuilder
 		nil,            // metricStmtBuilder
@@ -152,6 +154,7 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		metadataStore,
 		nil,              // prometheus
 		traceStmtBuilder, // traceStmtBuilder
+		nil,              // aiTraceStmtBuilder
 		nil,              // logStmtBuilder
 		nil,              // auditStmtBuilder
 		nil,              // metricStmtBuilder

pkg/querybuilder/filter_split.go

@@ -0,0 +1,152 @@
+package querybuilder
+
+import (
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/antlr4-go/antlr/v4"
+)
+
+// SplitFilterForAggregates partitions a single filter expression into a span-level
+// part (a WHERE over spans) and a trace-level part (a HAVING over per-trace
+// aggregates), splitting on the top-level AND.
+//
+// A key is trace-level when, with no explicit field context, its name is in
+// aggregateNames — written bare (`completion_tokens`) or with the user-facing `trace.`
+// prefix (`trace.completion_tokens`). Any explicit context (`span.`, `resource.`, …) is
+// span-level. Trace-level and span-level keys may be AND-combined (they run at different
+// query stages) but not OR-combined; an OR that mixes the two is an error.
+//
+// Syntax errors are ignored here — each part is re-parsed downstream (PrepareWhereClause
+// for the span part, the HAVING rewriter for the trace part), which surface them.
+func SplitFilterForAggregates(query string, aggregateNames map[string]struct{}) (spanExpr string, havingExpr string, err error) {
+	if strings.TrimSpace(query) == "" {
+		return "", "", nil
+	}
+
+	s := filterSplitter{query: query, aggregateNames: aggregateNames}
+	s.visit(parseFilterQuery(query))
+
+	if s.mixed {
+		return "", "", errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"trace-level and span-level filters cannot be combined within an OR/NOT group; separate them with a top-level AND")
+	}
+	return strings.Join(s.span, " AND "), strings.Join(s.having, " AND "), nil
+}
+
+func parseFilterQuery(query string) antlr.Tree {
+	lexer := grammar.NewFilterQueryLexer(antlr.NewInputStream(query))
+	lexer.RemoveErrorListeners()
+	parser := grammar.NewFilterQueryParser(antlr.NewCommonTokenStream(lexer, 0))
+	parser.RemoveErrorListeners()
+	return parser.Query()
+}
+
+// filterSplitter walks the parse tree once, flattening the top-level AND chain and
+// routing each atom (a comparison, a NOT expression, or a whole multi-branch OR group)
+// to the span or having bucket by the class of the keys it references.
+type filterSplitter struct {
+	query          string
+	aggregateNames map[string]struct{}
+	span           []string
+	having         []string
+	mixed          bool
+}
+
+func (s *filterSplitter) visit(node antlr.Tree) {
+	switch n := node.(type) {
+	case *grammar.QueryContext:
+		if n.Expression() != nil {
+			s.visit(n.Expression())
+		}
+	case *grammar.ExpressionContext:
+		if n.OrExpression() != nil {
+			s.visit(n.OrExpression())
+		}
+	case *grammar.OrExpressionContext:
+		// a single branch is just an AND chain; multiple branches are a real OR, kept
+		// whole so a class-mixing OR can be rejected.
+		if ands := n.AllAndExpression(); len(ands) == 1 {
+			s.visit(ands[0])
+		} else {
+			s.route(n)
+		}
+	case *grammar.AndExpressionContext:
+		for _, u := range n.AllUnaryExpression() {
+			s.visit(u)
+		}
+	case *grammar.UnaryExpressionContext:
+		if n.NOT() != nil {
+			s.route(n)
+		} else if n.Primary() != nil {
+			s.visit(n.Primary())
+		}
+	case *grammar.PrimaryContext:
+		if n.OrExpression() != nil { // parenthesized sub-expression
+			s.visit(n.OrExpression())
+		} else {
+			s.route(n)
+		}
+	}
+}
+
+// route classifies an atom and appends its original source text to the right bucket.
+func (s *filterSplitter) route(atom antlr.ParserRuleContext) {
+	isTrace, isSpan := classifyKeys(atom, s.aggregateNames)
+	if isTrace && isSpan {
+		s.mixed = true
+		return
+	}
+	text := atomSourceText(s.query, atom)
+	// A multi-branch OR group's source slice excludes its enclosing parens (they belong
+	// to the parent Primary). Re-wrap it so rejoining a bucket with " AND " cannot invert
+	// OR/AND precedence, e.g. `a AND (b OR c)` must not flatten to `a AND b OR c`.
+	if or, ok := atom.(*grammar.OrExpressionContext); ok && len(or.AllAndExpression()) > 1 {
+		text = "(" + text + ")"
+	}
+	if isTrace {
+		s.having = append(s.having, text)
+	} else {
+		s.span = append(s.span, text)
+	}
+}
+
+// classifyKeys reports whether a subtree references trace-level and/or span-level keys.
+// A key is trace-level only when it has no explicit field context and its name — after
+// the optional user-facing `trace.` prefix is stripped — is a known aggregate. Any
+// explicit context (`span.`, `resource.`, …) is span-level.
+func classifyKeys(node antlr.Tree, aggregateNames map[string]struct{}) (isTrace, isSpan bool) {
+	kc, ok := node.(*grammar.KeyContext)
+	if ok {
+		key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
+		if key.FieldContext == telemetrytypes.FieldContextUnspecified {
+			// `trace.` is the user-facing prefix for trace-level aggregates. It is not a
+			// registered field context, so it stays on the name; strip it before matching.
+			name := strings.TrimPrefix(key.Name, telemetrytypes.FieldContextTrace.StringValue()+".")
+			_, isTrace = aggregateNames[name]
+			isSpan = !isTrace
+		} else {
+			isSpan = true
+		}
+		return
+	}
+	for i := 0; i < node.GetChildCount(); i++ {
+		t, s := classifyKeys(node.GetChild(i), aggregateNames)
+		isTrace = isTrace || t
+		isSpan = isSpan || s
+	}
+	return
+}
+
+// atomSourceText returns the original source substring for an atom, preserving
+// whitespace. The token stream drops skipped whitespace, which would glue word
+// operators (OR/AND/NOT) to their operands, so slice the input by char offsets.
+func atomSourceText(query string, atom antlr.ParserRuleContext) string {
+	start, stop := atom.GetStart(), atom.GetStop()
+	if start == nil || stop == nil || start.GetStart() < 0 || stop.GetStop() >= len(query) || stop.GetStop() < start.GetStart() {
+		return atom.GetText()
+	}
+	return query[start.GetStart() : stop.GetStop()+1]
+}

pkg/querybuilder/filter_split_test.go

@@ -0,0 +1,156 @@
+package querybuilder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSplitFilterForAggregates(t *testing.T) {
+	agg := map[string]struct{}{"completion_tokens": {}, "span_count": {}, "prompt_tokens": {}}
+
+	type tc struct {
+		name    string
+		query   string
+		span    string // expected span-level (WHERE) part; "" => empty
+		having  string // expected trace-level (HAVING) part; "" => empty
+		wantErr bool
+	}
+
+	cases := []tc{
+		// --- empty input ---------------------------------------------------------
+		{
+			name: "empty",
+		},
+		{
+			name:  "whitespace only",
+			query: "   ",
+		},
+
+		// --- single class --------------------------------------------------------
+		{
+			name:  "span only",
+			query: "service.name = 'x'",
+			span:  "service.name = 'x'",
+		},
+		{
+			name:   "agg only bare",
+			query:  "completion_tokens > 1000",
+			having: "completion_tokens > 1000",
+		},
+		{
+			// the user-facing `trace.` prefix marks a trace-level aggregate.
+			name:   "agg only trace prefix",
+			query:  "trace.completion_tokens > 1000",
+			having: "trace.completion_tokens > 1000",
+		},
+		{
+			// `tracefield.` is not supported: it has an explicit context, so it is span-level.
+			name:  "tracefield prefix is span-level",
+			query: "tracefield.completion_tokens > 1000",
+			span:  "tracefield.completion_tokens > 1000",
+		},
+
+		// --- top-level AND splits across the two buckets -------------------------
+		{
+			name:   "span AND agg",
+			query:  "service.name = 'x' AND completion_tokens > 1000",
+			span:   "service.name = 'x'",
+			having: "completion_tokens > 1000",
+		},
+		{
+			// order within a bucket is preserved; the two span atoms join with AND.
+			name:   "span AND span AND agg",
+			query:  "service.name = 'x' AND kind_string = 'Internal' AND completion_tokens > 1000",
+			span:   "service.name = 'x' AND kind_string = 'Internal'",
+			having: "completion_tokens > 1000",
+		},
+		{
+			// a parenthesized top-level AND still splits across the two buckets.
+			name:   "parenthesized span AND agg",
+			query:  "(service.name = 'x' AND completion_tokens > 1000)",
+			span:   "service.name = 'x'",
+			having: "completion_tokens > 1000",
+		},
+
+		// --- OR groups are re-wrapped in parens so a later AND-join can't invert
+		//     precedence (`a AND (b OR c)` must not flatten to `a AND b OR c`) ------
+		{
+			name:   "agg OR agg",
+			query:  "completion_tokens > 1000 OR span_count > 3",
+			having: "(completion_tokens > 1000 OR span_count > 3)",
+		},
+		{
+			name:  "span OR span",
+			query: "service.name = 'x' OR kind_string = 'Internal'",
+			span:  "(service.name = 'x' OR kind_string = 'Internal')",
+		},
+		{
+			name:  "span AND (span OR span)",
+			query: "service.name = 'x' AND (kind_string = 'Internal' OR kind_string = 'Client')",
+			span:  "service.name = 'x' AND (kind_string = 'Internal' OR kind_string = 'Client')",
+		},
+		{
+			name:   "agg AND (agg OR agg)",
+			query:  "prompt_tokens > 5 AND (completion_tokens > 1000 OR span_count > 3)",
+			having: "prompt_tokens > 5 AND (completion_tokens > 1000 OR span_count > 3)",
+		},
+		{
+			// the OR group routes to span, the trailing aggregate to having.
+			name:   "span AND (span OR span) AND agg",
+			query:  "a.b = 'x' AND (c.d = 'y' OR e.f = 'z') AND completion_tokens > 1000",
+			span:   "a.b = 'x' AND (c.d = 'y' OR e.f = 'z')",
+			having: "completion_tokens > 1000",
+		},
+
+		// --- a nested AND group flattens across the buckets (no spurious parens) --
+		{
+			name:   "(span AND agg) AND agg",
+			query:  "(service.name = 'x' AND completion_tokens > 1000) AND prompt_tokens > 5",
+			span:   "service.name = 'x'",
+			having: "completion_tokens > 1000 AND prompt_tokens > 5",
+		},
+
+		// --- NOT wrapping a single-class group is routed whole to that class ------
+		{
+			name:   "not agg",
+			query:  "NOT (completion_tokens > 1000)",
+			having: "NOT (completion_tokens > 1000)",
+		},
+		{
+			name:  "not span",
+			query: "NOT (service.name = 'x')",
+			span:  "NOT (service.name = 'x')",
+		},
+
+		// --- class-mixing is rejected in an OR group, a NOT group, or a nested OR -
+		{
+			name:    "agg OR span rejected",
+			query:   "completion_tokens > 1000 OR service.name = 'x'",
+			wantErr: true,
+		},
+		{
+			name:    "not mixed rejected",
+			query:   "NOT (completion_tokens > 1000 AND service.name = 'x')",
+			wantErr: true,
+		},
+		{
+			name:    "span AND (agg OR span) rejected",
+			query:   "service.name = 'x' AND (completion_tokens > 1000 OR kind_string = 'Client')",
+			wantErr: true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			span, having, err := SplitFilterForAggregates(c.query, agg)
+			if c.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, c.span, span, "span part")
+			require.Equal(t, c.having, having, "having part")
+		})
+	}
+}

pkg/querybuilder/having_rewriter.go

@@ -18,6 +18,19 @@ func NewHavingExpressionRewriter() *HavingExpressionRewriter {
 	}
 }
 
+// Rewrite rewrites and validates a HAVING expression against a caller-supplied
+// column map (user-facing name -> SQL identifier/expression). Values are inlined, so
+// the result is a bare SQL boolean expression with no bound args. Used by callers
+// that project their own aggregate columns (e.g. the AI trace list) rather than the
+// query's Aggregations.
+func (r *HavingExpressionRewriter) Rewrite(expression string, columnMap map[string]string) (string, error) {
+	if len(strings.TrimSpace(expression)) == 0 {
+		return "", nil
+	}
+	r.columnMap = columnMap
+	return r.rewriteAndValidate(expression)
+}
+
 // RewriteForTraces rewrites and validates the HAVING expression for a traces query.
 func (r *HavingExpressionRewriter) RewriteForTraces(expression string, aggregations []qbtypes.TraceAggregation) (string, error) {
 	if len(strings.TrimSpace(expression)) == 0 {

pkg/telemetryai/base_condition.go

@@ -0,0 +1,55 @@
+package telemetryai
+
+import (
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+// BaseConditionProvider defines which spans (and traces) are in scope, decoupled
+// from the query. Swap it to redefine "an AI trace" without touching the builder.
+//
+// It only *declares* the gate (a grammar expression + its field keys); the builder
+// resolves those keys through the field mapper, so all attribute access is
+// materialization/evolution aware — no hardcoded map lookups.
+type BaseConditionProvider interface {
+	// FilterExpression is the grammar-level (EXISTS) gate, resolved via the visitor.
+	FilterExpression() string
+	// FieldKeys are the gate's keys: registered in metadata and used to build the
+	// per-span mask (OR of resolved EXISTS conditions).
+	FieldKeys() []*telemetrytypes.TelemetryFieldKey
+}
+
+// genAIBaseConditionProvider: an AI trace has >=1 gen_ai LLM, tool, or agent span.
+type genAIBaseConditionProvider struct {
+	keys []string
+}
+
+var _ BaseConditionProvider = (*genAIBaseConditionProvider)(nil)
+
+func NewGenAIBaseConditionProvider() BaseConditionProvider {
+	return &genAIBaseConditionProvider{
+		keys: []string{telemetrytypes.GenAIRequestModel, telemetrytypes.GenAIToolName, telemetrytypes.GenAIAgentName},
+	}
+}
+
+func (p *genAIBaseConditionProvider) FilterExpression() string {
+	parts := make([]string, 0, len(p.keys))
+	for _, k := range p.keys {
+		parts = append(parts, k+" EXISTS")
+	}
+	return strings.Join(parts, " OR ")
+}
+
+func (p *genAIBaseConditionProvider) FieldKeys() []*telemetrytypes.TelemetryFieldKey {
+	keys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(p.keys))
+	for _, k := range p.keys {
+		keys = append(keys, &telemetrytypes.TelemetryFieldKey{
+			Name:          k,
+			Signal:        telemetrytypes.SignalTraces,
+			FieldContext:  telemetrytypes.FieldContextAttribute,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		})
+	}
+	return keys
+}

pkg/telemetryai/projection.go

@@ -0,0 +1,99 @@
+package telemetryai
+
+import (
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+type TraceColumn struct {
+	Alias string
+	// Orderable marks a column computable in the mask-pruned `matched` pass (a
+	// gen_ai-scoped aggregate): the only columns usable for ORDER BY and the
+	// aggregate filter. All-span columns (span_count, duration_nano, …) are false —
+	// they are output-only, since the matched pass sees only gen_ai spans.
+	Orderable bool
+	Intrinsic string   // fixed SQL over intrinsic columns; used verbatim (nil Attr)
+	Attr      *AttrAgg // field-mapper-resolved aggregate (nil Intrinsic)
+}
+
+// AttrAgg describes an aggregate the builder assembles after resolving fields:
+//   - Func "count" + ExistsKey  -> countIf(<ExistsKey EXISTS>)   (e.g. llm_call_count)
+//   - Func + ValueKey           -> Func(<resolved value>)         (e.g. sum tokens)
+//   - Func + ValueExpr + Scoped -> FuncIf(<value>, <gate mask>)   (e.g. last_activity_time)
+type AttrAgg struct {
+	Func      string                            // sum | max | min | count
+	ValueKey  *telemetrytypes.TelemetryFieldKey // attribute value to aggregate (resolved as Float64)
+	ValueExpr string                            // fixed value expr when ValueKey is nil (e.g. "timestamp")
+	Scoped    bool                              // wrap as <Func>If(value, <gate mask>)
+	ExistsKey *telemetrytypes.TelemetryFieldKey // count spans where this key exists (countIf)
+}
+
+// ProjectionProvider decides which per-trace columns the list computes and which
+// are sortable, decoupled from selection and topology.
+type ProjectionProvider interface {
+	Columns() []TraceColumn
+	// DefaultOrderAlias is sorted by (desc) when the query gives no order.
+	DefaultOrderAlias() string
+	// AggregateAliases are the computed per-trace (trace-level) column names, used to
+	// classify which filter keys are trace-level vs span-level. Only the Orderable
+	// subset is actually usable in ORDER BY / the aggregate filter; the rest are
+	// output-only. Excludes aliases that are also real span/resource keys (service.name).
+	AggregateAliases() []string
+}
+
+// CommonTraceColumns are domain-neutral intrinsic columns any trace list can reuse.
+// All are over-all-spans intrinsics, so none is Orderable (not computable in the
+// mask-pruned matched pass) — they are output-only, computed in the enrichment scan.
+func CommonTraceColumns() []TraceColumn {
+	return []TraceColumn{
+		{Alias: "start_time", Intrinsic: "min(timestamp)", Orderable: false},
+		{Alias: "end_time", Intrinsic: "max(timestamp)", Orderable: false},
+		{Alias: "duration_nano", Intrinsic: "(max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp)))", Orderable: false},
+		{Alias: "span_count", Intrinsic: "count()", Orderable: false},
+
+		{Alias: "root_span_name", Intrinsic: "anyIf(name, parent_span_id = '')", Orderable: false},
+		{Alias: "service.name", Intrinsic: "any(resource_string_service$$name)", Orderable: false},
+	}
+}
+
+// genAIProjectionProvider adds AI/LLM per-trace metrics to the common columns.
+type genAIProjectionProvider struct{}
+
+var _ ProjectionProvider = (*genAIProjectionProvider)(nil)
+
+func NewGenAIProjectionProvider() ProjectionProvider {
+	return &genAIProjectionProvider{}
+}
+
+func (genAIProjectionProvider) Columns() []TraceColumn {
+	// Key definitions (name/context/type) live once in GenAIFieldDefinitions, shared
+	// with the metadata enrichment. Copy to locals so we can take their address.
+	reqModel := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIRequestModel]
+	inTok := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIUsageInputTokens]
+	outTok := telemetrytypes.GenAIFieldDefinitions[telemetrytypes.GenAIUsageOutputTokens]
+
+	cols := CommonTraceColumns()
+	return append(cols,
+		// LLM calls only (request model present), not the full gate (tool/agent too).
+		TraceColumn{Alias: "llm_call_count", Orderable: true, Attr: &AttrAgg{Func: "count", ExistsKey: &reqModel}},
+		// tokens live only on LLM spans, so a plain sum over resolved values (NULL
+		// elsewhere) is correct without scoping to the mask. Aliases match the OTel
+		// source attributes (gen_ai.usage.input_tokens / output_tokens).
+		TraceColumn{Alias: "input_tokens", Orderable: true, Attr: &AttrAgg{Func: "sum", ValueKey: &inTok}},
+		TraceColumn{Alias: "output_tokens", Orderable: true, Attr: &AttrAgg{Func: "sum", ValueKey: &outTok}},
+		// timestamp of the last in-scope (gen_ai: LLM/tool/agent) span.
+		TraceColumn{Alias: "last_activity_time", Orderable: true, Attr: &AttrAgg{Func: "max", ValueExpr: "timestamp", Scoped: true}},
+	)
+}
+
+func (genAIProjectionProvider) DefaultOrderAlias() string { return "last_activity_time" }
+
+func (genAIProjectionProvider) AggregateAliases() []string {
+	// every computed column except service.name (a real resource key, filterable
+	// at the span level). Of these, only the gen_ai-scoped ones (llm_call_count,
+	// input_tokens, output_tokens, last_activity_time) are Orderable and thus usable
+	// in ORDER BY / the aggregate filter; the rest are output-only.
+	return []string{
+		"start_time", "end_time", "duration_nano", "span_count", "root_span_name",
+		"llm_call_count", "input_tokens", "output_tokens", "last_activity_time",
+	}
+}

pkg/telemetryai/statement_builder.go

@@ -0,0 +1,722 @@
+package telemetryai
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetrytraces"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+var (
+	ErrUnsupportedRequestType = errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type for source=ai")
+)
+
+// scopedTraceStatementBuilder builds a trace list scoped to a span-selection
+// category. Topology is fixed; selection (BaseConditionProvider) and columns
+// (ProjectionProvider) are pluggable, so a new category is a new pair of
+// providers, not new topology.
+type scopedTraceStatementBuilder struct {
+	logger           *slog.Logger
+	metadataStore    telemetrytypes.MetadataStore
+	fm               qbtypes.FieldMapper
+	cb               qbtypes.ConditionBuilder
+	baseCond         BaseConditionProvider
+	projection       ProjectionProvider
+	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
+}
+
+var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*scopedTraceStatementBuilder)(nil)
+
+// NewScopedTraceStatementBuilder wires the generic trace-list builder. The trace
+// builder is reused for the span-list (raw) path.
+func NewScopedTraceStatementBuilder(
+	settings factory.ProviderSettings,
+	metadataStore telemetrytypes.MetadataStore,
+	fieldMapper qbtypes.FieldMapper,
+	conditionBuilder qbtypes.ConditionBuilder,
+	baseCond BaseConditionProvider,
+	projection ProjectionProvider,
+	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
+) *scopedTraceStatementBuilder {
+	aiSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryai")
+	return &scopedTraceStatementBuilder{
+		logger:           aiSettings.Logger(),
+		metadataStore:    metadataStore,
+		fm:               fieldMapper,
+		cb:               conditionBuilder,
+		baseCond:         baseCond,
+		projection:       projection,
+		traceStmtBuilder: traceStmtBuilder,
+	}
+}
+
+// NewAITraceStatementBuilder is the scoped builder with the gen_ai gate + AI projection.
+func NewAITraceStatementBuilder(
+	settings factory.ProviderSettings,
+	metadataStore telemetrytypes.MetadataStore,
+	fieldMapper qbtypes.FieldMapper,
+	conditionBuilder qbtypes.ConditionBuilder,
+	baseCond BaseConditionProvider,
+	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
+) *scopedTraceStatementBuilder {
+	return NewScopedTraceStatementBuilder(settings, metadataStore, fieldMapper, conditionBuilder, baseCond, NewGenAIProjectionProvider(), traceStmtBuilder)
+}
+
+func (b *scopedTraceStatementBuilder) Build(
+	ctx context.Context,
+	start uint64,
+	end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	switch requestType {
+	case qbtypes.RequestTypeTrace:
+		return b.buildTraceListQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
+	case qbtypes.RequestTypeRaw:
+		return b.buildDelegated(ctx, start, end, requestType, query, variables)
+	default:
+		return nil, ErrUnsupportedRequestType
+	}
+}
+
+// buildDelegated ANDs the base gate into the user filter and delegates to the
+// standard trace builder (the span-list / raw path).
+func (b *scopedTraceStatementBuilder) buildDelegated(
+	ctx context.Context,
+	start, end uint64,
+	requestType qbtypes.RequestType,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+	gate := b.baseCond.FilterExpression()
+	expr := gate
+	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
+		expr = fmt.Sprintf("(%s) AND (%s)", gate, query.Filter.Expression)
+	}
+
+	// shallow copy; only Filter is replaced, caller's query untouched
+	gated := query
+	gated.Filter = &qbtypes.Filter{Expression: expr}
+
+	return b.traceStmtBuilder.Build(ctx, start, end, requestType, gated, variables)
+}
+
+// buildTraceListQuery is the map for the whole file. It resolves the columns, then
+// wires the CTE pipeline that was benchmarked (see ai-qb-handoff.md): a single
+// windowed pass picks the top-N traces, then a bucket-pruned pass enriches only those.
+// The helpers appear in this file in the order they run here.
+//
+//	RESOLVE (turn keys/columns into SQL, field-mapper aware)
+//	  fetchKeys        → metadata for the keys we reference
+//	  resolveMask      → the "is a gen_ai span" predicate  (OR of EXISTS)  [existsExpr]
+//	  resolveColumns   → per-trace column SQL: intrinsics + resolved aggregates
+//	  resolveListOrders→ which resolved columns to ORDER BY
+//	  splitFilter      → span-level predicate + trace-level HAVING expression
+//
+//	BUILD (compose the CTE pipeline)
+//	  matched   [buildMatchedCTE]   ONE windowed, mask-pruned GROUP BY trace_id pass over
+//	     │                          the span index that applies the gate (+ span filter as
+//	     │                          countIf existence), the trace-level HAVING, ORDER BY and
+//	     │                          LIMIT/OFFSET in a single scan → top-N trace_ids + their
+//	     │                          gen_ai-scoped ranking metrics. No giant gate id-set.
+//	     ▼
+//	  ranked    [buildRankedCTE]    per-trace [start,end] bounds for those N traces, read
+//	     │                          from the small distributed_trace_summary table.
+//	     ▼
+//	  buckets   [buildBucketsCTE]   the exact ts_bucket_start values those N traces touch,
+//	     │                          so the enrichment scan is primary-key pruned.
+//	     ▼
+//	  enrichment[buildEnrichmentSelect]  all per-trace columns for the N traces, scanning
+//	                                     only their buckets (full trace, not window-clipped).
+//
+// Only gen_ai-scoped aggregates (tokens, llm activity, llm_call_count) are computable in
+// the mask-pruned `matched` pass, so only those are orderable / usable in the aggregate
+// filter. All-span columns (span_count, duration_nano, …) are output-only.
+//
+// start/end are nanoseconds.
+func (b *scopedTraceStatementBuilder) buildTraceListQuery(
+	ctx context.Context,
+	start, end uint64,
+	query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
+	variables map[string]qbtypes.VariableItem,
+) (*qbtypes.Statement, error) {
+
+	startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
+	endBucket := end / querybuilder.NsToSeconds
+
+	limit := query.Limit
+	if limit <= 0 {
+		limit = 100
+	}
+
+	// Resolve the gate keys + columns once; every attribute access below goes through
+	// the field mapper (materialization/evolution aware), never a hardcoded map lookup.
+	keys, err := b.fetchKeys(ctx)
+	if err != nil {
+		return nil, err
+	}
+	maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
+	if err != nil {
+		return nil, err
+	}
+	resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
+	if err != nil {
+		return nil, err
+	}
+	orders, err := b.resolveListOrders(query.Order, resolved)
+	if err != nil {
+		return nil, err
+	}
+	orderableSet := orderableAliasSet(resolved)
+
+	// Split the user filter: span-level predicate + trace-level HAVING expression.
+	fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), orderableSet, start, end, variables)
+	if err != nil {
+		return nil, err
+	}
+
+	// matched → ranked → buckets → enrichment
+	matchedFrag, matchedArgs, err := b.buildMatchedCTE(start, end, startBucket, endBucket, resolved, orders, orderableSet, maskExpr, maskArgs, fp, limit, query.Offset)
+	if err != nil {
+		return nil, err
+	}
+	rankedFrag, rankedArgs := b.buildRankedCTE(start, end)
+	bucketsFrag := buildBucketsCTE()
+	mainSQL, mainArgs := b.buildEnrichmentSelect(resolved, orders)
+
+	cteFragments := []string{matchedFrag, rankedFrag, bucketsFrag}
+	cteArgs := [][]any{matchedArgs, rankedArgs, nil}
+
+	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
+	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
+
+	return &qbtypes.Statement{
+		Query:          finalSQL,
+		Args:           finalArgs,
+		Warnings:       fp.warnings,
+		WarningsDocURL: fp.warningsURL,
+	}, nil
+}
+
+// ---------------------------------------------------------------------------
+// RESOLVE — turn keys/columns into field-mapper-aware SQL
+// ---------------------------------------------------------------------------
+
+func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
+	fields := b.resolverFieldKeys()
+	selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields))
+	for _, k := range fields {
+		selectors = append(selectors, &telemetrytypes.FieldKeySelector{
+			Name:         k.Name,
+			Signal:       k.Signal,
+			FieldContext: k.FieldContext,
+		})
+	}
+	keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
+	return keys, err
+}
+
+func (b *scopedTraceStatementBuilder) resolverFieldKeys() []*telemetrytypes.TelemetryFieldKey {
+	seen := make(map[string]struct{})
+	var out []*telemetrytypes.TelemetryFieldKey
+	add := func(k *telemetrytypes.TelemetryFieldKey) {
+		if k == nil {
+			return
+		}
+		if _, dup := seen[k.Name]; dup {
+			return
+		}
+		seen[k.Name] = struct{}{}
+		out = append(out, k)
+	}
+	for _, k := range b.baseCond.FieldKeys() {
+		add(k)
+	}
+	for _, c := range b.projection.Columns() {
+		if c.Attr != nil {
+			add(c.Attr.ValueKey)
+			add(c.Attr.ExistsKey)
+		}
+	}
+	return out
+}
+
+// resolveMask builds the per-span in-scope mask: OR of resolved EXISTS predicates
+// over the base condition's field keys.
+func (b *scopedTraceStatementBuilder) resolveMask(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey) (string, []any, error) {
+	fieldKeys := b.baseCond.FieldKeys()
+	parts := make([]string, 0, len(fieldKeys))
+	var args []any
+	for _, key := range fieldKeys {
+		e, a, err := b.existsExpr(ctx, start, end, keys, key)
+		if err != nil {
+			return "", nil, err
+		}
+		parts = append(parts, e)
+		args = append(args, a...)
+	}
+	return "(" + strings.Join(parts, " OR ") + ")", args, nil
+}
+
+// existsExpr resolves a field-mapper-aware EXISTS predicate for key (materialized
+// column when present, else the map). Escaped once so it round-trips when embedded
+// in an outer builder.
+func (b *scopedTraceStatementBuilder) existsExpr(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, key *telemetrytypes.TelemetryFieldKey) (string, []any, error) {
+	resolvedKey := key
+	cands := keys[key.Name]
+	if len(cands) == 0 {
+		cands = []*telemetrytypes.TelemetryFieldKey{key}
+	} else {
+		resolvedKey = cands[0]
+	}
+	sb := sqlbuilder.NewSelectBuilder()
+	conds, _, err := b.cb.ConditionFor(ctx, start, end, resolvedKey, cands, qbtypes.FilterOperatorExists, nil, sb)
+	if err != nil {
+		return "", nil, err
+	}
+	sb.Where(conds[0])
+	expr, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	expr = strings.TrimPrefix(expr, "WHERE ")
+	return sqlbuilder.Escape(expr), args, nil
+}
+
+// resolvedColumn is a projection column whose attribute access has been resolved to
+// SQL via the field mapper. expr is escaped once, ready to embed in an outer SELECT.
+type resolvedColumn struct {
+	alias     string
+	expr      string
+	args      []any
+	orderable bool
+}
+
+// resolveColumns turns the projection's declarative columns into SQL, resolving all
+// attribute access through the field mapper.
+func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, start, end uint64, keys map[string][]*telemetrytypes.TelemetryFieldKey, maskExpr string, maskArgs []any) ([]resolvedColumn, error) {
+	cols := b.projection.Columns()
+	out := make([]resolvedColumn, 0, len(cols))
+	for _, c := range cols {
+		rc := resolvedColumn{alias: c.Alias, orderable: c.Orderable}
+
+		if c.Attr == nil {
+			// intrinsic: escape once (no-op unless it references a $$ column).
+			rc.expr = sqlbuilder.Escape(c.Intrinsic)
+			out = append(out, rc)
+			continue
+		}
+
+		a := c.Attr
+		switch {
+		case a.Func == "count" && a.ExistsKey != nil:
+			cond, cargs, err := b.existsExpr(ctx, start, end, keys, a.ExistsKey)
+			if err != nil {
+				return nil, err
+			}
+			rc.expr = fmt.Sprintf("countIf(%s)", cond)
+			rc.args = cargs
+		default:
+			var vexpr string
+			var vargs []any
+			if a.ValueKey != nil {
+				e, ar, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, a.ValueKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeFloat64, nil, false)
+				if err != nil {
+					return nil, err
+				}
+				vexpr, vargs = e, ar
+			} else {
+				vexpr = a.ValueExpr
+			}
+			if a.Scoped {
+				rc.expr = fmt.Sprintf("%sIf(%s, %s)", a.Func, vexpr, maskExpr)
+				rc.args = append(append([]any{}, vargs...), maskArgs...)
+			} else {
+				rc.expr = fmt.Sprintf("%s(%s)", a.Func, vexpr)
+				rc.args = vargs
+			}
+		}
+		out = append(out, rc)
+	}
+	return out, nil
+}
+
+// listOrder resolves a sort key to an aggregate-column alias + direction. Both the
+// matched CTE and the enrichment select that alias, so both ORDER BY it.
+type listOrder struct {
+	alias     string
+	direction string
+}
+
+// resolveListOrders maps order keys to the resolved orderable columns; non-orderable
+// columns are rejected. Defaults to the projection's default order.
+func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy, resolved []resolvedColumn) ([]listOrder, error) {
+	byAlias := make(map[string]resolvedColumn, len(resolved))
+	orderable := make([]string, 0, len(resolved))
+	for _, rc := range resolved {
+		byAlias[rc.alias] = rc
+		if rc.orderable {
+			orderable = append(orderable, rc.alias)
+		}
+	}
+
+	if len(order) == 0 {
+		return []listOrder{{alias: b.projection.DefaultOrderAlias(), direction: "DESC"}}, nil
+	}
+
+	orders := make([]listOrder, 0, len(order))
+	for _, o := range order {
+		direction := "DESC"
+		if o.Direction == qbtypes.OrderDirectionAsc {
+			direction = "ASC"
+		}
+		rc, ok := byAlias[o.Key.Name]
+		if !ok || !rc.orderable {
+			return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"unsupported order key %q for the trace list; orderable keys: %s", o.Key.Name, strings.Join(orderable, ", "))
+		}
+		orders = append(orders, listOrder{alias: rc.alias, direction: direction})
+	}
+	return orders, nil
+}
+
+// filterParts is the split of the user filter into a resolved span-level predicate
+// (used both to widen the matched WHERE prune and as a countIf existence in HAVING)
+// and a trace-level HAVING expression.
+type filterParts struct {
+	spanPred      string
+	spanArgs      []any
+	hasSpanFilter bool
+	havingExpr    string
+	warnings      []string
+	warningsURL   string
+}
+
+// splitFilter partitions query.Filter into a span-level predicate and a trace-level
+// HAVING expression; an explicit query.Having is ANDed onto the latter. The trace-level
+// expression is validated against the aggregates computable in the matched pass.
+func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, orderableSet map[string]struct{}, start, end uint64, variables map[string]qbtypes.VariableItem) (filterParts, error) {
+	var fp filterParts
+	if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
+		spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
+		if err != nil {
+			return fp, err
+		}
+		fp.havingExpr = traceExpr
+		if strings.TrimSpace(spanExpr) != "" {
+			pred, args, warnings, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, variables)
+			if err != nil {
+				return fp, err
+			}
+			fp.spanPred, fp.spanArgs, fp.hasSpanFilter = pred, args, true
+			fp.warnings, fp.warningsURL = warnings, url
+		}
+	}
+	if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
+		if fp.havingExpr != "" {
+			fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
+		} else {
+			fp.havingExpr = query.Having.Expression
+		}
+	}
+	if err := validateAggregateFilter(fp.havingExpr, orderableSet); err != nil {
+		return fp, err
+	}
+	return fp, nil
+}
+
+// resolveSpanPredicate resolves a span-level filter expression to a bare boolean SQL
+// predicate (escaped) + args via the field mapper.
+func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, start, end uint64, expr string, variables map[string]qbtypes.VariableItem) (string, []any, []string, string, error) {
+	selectors := querybuilder.QueryStringToKeysSelectors(expr)
+	for i := range selectors {
+		selectors[i].Signal = telemetrytypes.SignalTraces
+	}
+	keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
+	if err != nil {
+		return "", nil, nil, "", err
+	}
+	prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
+		Context:          ctx,
+		Logger:           b.logger,
+		FieldMapper:      b.fm,
+		ConditionBuilder: b.cb,
+		FieldKeys:        keys,
+		Variables:        variables,
+		StartNs:          start,
+		EndNs:            end,
+	})
+	if err != nil {
+		return "", nil, nil, "", err
+	}
+	if prepared.IsEmpty() {
+		return "", nil, nil, "", nil
+	}
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("1")
+	sb.AddWhereClause(prepared.WhereClause)
+	sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	pred := sql[strings.Index(sql, "WHERE ")+len("WHERE "):]
+	return sqlbuilder.Escape(pred), args, prepared.Warnings, prepared.WarningsDocURL, nil
+}
+
+// ---------------------------------------------------------------------------
+// BUILD — compose the CTE pipeline (matched → ranked → buckets → enrichment)
+// ---------------------------------------------------------------------------
+
+// buildMatchedCTE builds `matched`: a single windowed GROUP BY trace_id pass that
+// picks the top-N traces. The WHERE prunes to gen_ai spans (widened with the span-level
+// predicate when present); HAVING enforces the gate (countIf(mask) > 0), the span-level
+// filter as an existence check, and the trace-level aggregate filter; ORDER BY + LIMIT
+// select the winners. Only gen_ai-scoped aggregates are computable here, and only those
+// actually referenced by ORDER BY / the aggregate filter are selected (the rest are
+// computed once, later, in the enrichment scan) — this keeps the hot ranking pass lean.
+func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet map[string]struct{}, maskExpr string, maskArgs []any, fp filterParts, limit, offset int) (string, []any, error) {
+	sb := sqlbuilder.NewSelectBuilder()
+
+	// SELECT trace_id + only the aggregates ORDER BY / HAVING reference (as aliases).
+	needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
+	selects := []string{"trace_id"}
+	for _, rc := range resolved {
+		if _, ok := needed[rc.alias]; !ok {
+			continue
+		}
+		selects = append(selects, embedExpr(sb, rc.expr, rc.args)+" AS "+quoteAlias(rc.alias))
+	}
+	sb.Select(selects...)
+	sb.From(spanTable())
+
+	// WHERE: window + coarse prune to gen_ai spans (widened so span-filter spans are
+	// visible for the countIf existence check below).
+	win := windowWhere(sb, start, end, startBucket, endBucket)
+	prune := "(" + embedExpr(sb, maskExpr, maskArgs)
+	if fp.hasSpanFilter {
+		prune += " OR " + embedExpr(sb, fp.spanPred, fp.spanArgs)
+	}
+	prune += ")"
+	sb.Where(append(win, prune)...)
+	sb.GroupBy("trace_id")
+
+	// HAVING: the gate + span-existence checks are only needed once the WHERE has been
+	// widened by a span filter; otherwise WHERE = mask already enforces the gate.
+	var having []string
+	if fp.hasSpanFilter {
+		having = append(having, "countIf("+embedExpr(sb, maskExpr, maskArgs)+") > 0")
+		having = append(having, "countIf("+embedExpr(sb, fp.spanPred, fp.spanArgs)+") > 0")
+	}
+	if strings.TrimSpace(fp.havingExpr) != "" {
+		hv, err := b.buildHaving(fp.havingExpr, orderableSet)
+		if err != nil {
+			return "", nil, err
+		}
+		if hv != "" {
+			having = append(having, hv)
+		}
+	}
+	if len(having) > 0 {
+		sb.Having(strings.Join(having, " AND "))
+	}
+
+	sb.OrderBy(orderClause(orders)...)
+	sb.Limit(limit)
+	if offset > 0 {
+		sb.Offset(offset)
+	}
+
+	sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return fmt.Sprintf("matched AS (%s)", sql), args, nil
+}
+
+// buildRankedCTE builds `ranked`: per-trace [start,end] bounds for the matched traces,
+// read from the small trace-summary table (used to derive the bucket prune).
+func (b *scopedTraceStatementBuilder) buildRankedCTE(start, end uint64) (string, []any) {
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("trace_id", "min(start) AS t_start", "max(end) AS t_end")
+	sb.From(summaryTable())
+	sb.Where(
+		"trace_id GLOBAL IN (SELECT trace_id FROM matched)",
+		"end >= fromUnixTimestamp64Nano("+sb.Var(start)+")",
+		"start < fromUnixTimestamp64Nano("+sb.Var(end)+")",
+	)
+	sb.GroupBy("trace_id")
+	sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return fmt.Sprintf("ranked AS (%s)", sql), args
+}
+
+// buildBucketsCTE builds `buckets`: the exact ts_bucket_start values the matched traces
+// span, so the enrichment scan is pruned to those primary-key buckets. No args.
+func buildBucketsCTE() string {
+	adj := querybuilder.BucketAdjustment // 30-min bucket width in seconds
+	return fmt.Sprintf("buckets AS (SELECT DISTINCT b AS ts_bucket FROM ranked "+
+		"ARRAY JOIN range("+
+		"toUInt64(intDiv(toUnixTimestamp(t_start), %d) * %d - %d), "+
+		"toUInt64(intDiv(toUnixTimestamp(t_end), %d) * %d + %d), "+
+		"%d) AS b)", adj, adj, adj, adj, adj, adj, adj)
+}
+
+// buildEnrichmentSelect builds the final SELECT: all per-trace columns for the matched
+// traces, scanning only their buckets (full trace, not window-clipped). SELECT-expr
+// args lead; the WHERE / ORDER BY carry none.
+func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(resolved []resolvedColumn, orders []listOrder) (string, []any) {
+	sb := sqlbuilder.NewSelectBuilder()
+	selects, selectArgs := selectAllColumns(resolved)
+	sb.Select(selects...)
+	sb.From(spanTable())
+	sb.Where(
+		"ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)",
+		"trace_id GLOBAL IN (SELECT trace_id FROM ranked)",
+	)
+	sb.GroupBy("trace_id")
+	sb.OrderBy(orderClause(orders)...)
+	sql, builtArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return sql, append(append([]any{}, selectArgs...), builtArgs...)
+}
+
+// buildHaving rewrites a trace-level HAVING expression against the aggregate column
+// aliases computable in the matched pass; bare, trace., and tracefield. forms all map
+// to the selected alias.
+func (b *scopedTraceStatementBuilder) buildHaving(havingExpr string, orderableSet map[string]struct{}) (string, error) {
+	columnMap := make(map[string]string, len(orderableSet)*3)
+	for a := range orderableSet {
+		columnMap[a] = quoteAlias(a)
+		columnMap["trace."+a] = quoteAlias(a)
+		columnMap["tracefield."+a] = quoteAlias(a)
+	}
+	return querybuilder.NewHavingExpressionRewriter().Rewrite(havingExpr, columnMap)
+}
+
+// ---------------------------------------------------------------------------
+// Small shared SQL-builder utilities
+// ---------------------------------------------------------------------------
+
+// spanTable is the fully-qualified span index table.
+func spanTable() string {
+	return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.SpanIndexV3TableName)
+}
+
+// summaryTable is the fully-qualified trace-summary table.
+func summaryTable() string {
+	return fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.TraceSummaryTableName)
+}
+
+// aggregateAliasSet is the set of all trace-level (computed) column aliases, used to
+// classify which filter keys are trace-level vs span-level.
+func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
+	set := make(map[string]struct{}, len(b.projection.AggregateAliases()))
+	for _, a := range b.projection.AggregateAliases() {
+		set[a] = struct{}{}
+	}
+	return set
+}
+
+// orderableAliasSet is the subset of aggregate aliases computable in the matched pass
+// (gen_ai-scoped): the only ones usable for ORDER BY and the aggregate filter.
+func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
+	set := make(map[string]struct{})
+	for _, rc := range resolved {
+		if rc.orderable {
+			set[rc.alias] = struct{}{}
+		}
+	}
+	return set
+}
+
+// neededMatchedAliases is the minimal set of aggregate aliases the matched pass must
+// select: those referenced by ORDER BY plus those referenced in the aggregate HAVING
+// (bare / trace. / tracefield. forms). Anything else is left to the enrichment scan.
+func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
+	needed := make(map[string]struct{})
+	for _, o := range orders {
+		needed[o.alias] = struct{}{}
+	}
+	for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
+		name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
+		if _, ok := orderableSet[name]; ok {
+			needed[name] = struct{}{}
+		}
+	}
+	return needed
+}
+
+// validateAggregateFilter rejects a trace-level filter that references an aggregate not
+// computable in the matched pass (e.g. span_count, duration_nano), with a clear message.
+func validateAggregateFilter(havingExpr string, orderableSet map[string]struct{}) error {
+	if strings.TrimSpace(havingExpr) == "" {
+		return nil
+	}
+	allowed := make([]string, 0, len(orderableSet))
+	for a := range orderableSet {
+		allowed = append(allowed, a)
+	}
+	for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
+		name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
+		if _, ok := orderableSet[name]; !ok {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"aggregate %q cannot be used in an AI trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
+		}
+	}
+	return nil
+}
+
+// embedExpr inlines a resolved (escaped) expr carrying `?` placeholders into sb by
+// replacing each `?` with a builder Var, so go-sqlbuilder tracks the args in appearance
+// order and un-escapes the expr at Build time.
+func embedExpr(sb *sqlbuilder.SelectBuilder, expr string, args []any) string {
+	var out strings.Builder
+	ai := 0
+	for i := 0; i < len(expr); i++ {
+		if expr[i] == '?' && ai < len(args) {
+			out.WriteString(sb.Var(args[ai]))
+			ai++
+			continue
+		}
+		out.WriteByte(expr[i])
+	}
+	return out.String()
+}
+
+// windowWhere binds the shared time-window predicates to sb and returns them, so a
+// caller can add its own predicate in the same Where call.
+func windowWhere(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64) []string {
+	return []string{
+		sb.GE("timestamp", fmt.Sprintf("%d", start)),
+		sb.L("timestamp", fmt.Sprintf("%d", end)),
+		sb.GE("ts_bucket_start", startBucket),
+		sb.LE("ts_bucket_start", endBucket),
+	}
+}
+
+// orderClause renders the ORDER BY terms (by column alias) + the trace_id tiebreak.
+func orderClause(orders []listOrder) []string {
+	out := make([]string, 0, len(orders)+1)
+	for _, o := range orders {
+		out = append(out, fmt.Sprintf("%s %s", quoteAlias(o.alias), o.direction))
+	}
+	return append(out, "trace_id DESC")
+}
+
+// selectAllColumns renders `expr AS alias` for every resolved column and returns their
+// field-mapper args in select order.
+func selectAllColumns(resolved []resolvedColumn) ([]string, []any) {
+	selects := []string{"trace_id"}
+	var args []any
+	for _, rc := range resolved {
+		selects = append(selects, rc.expr+" AS "+quoteAlias(rc.alias))
+		args = append(args, rc.args...)
+	}
+	return selects, args
+}
+
+// quoteAlias backticks an alias that carries characters special to the SQL builder.
+func quoteAlias(alias string) string {
+	if strings.ContainsAny(alias, ".$`") {
+		return "`" + alias + "`"
+	}
+	return alias
+}

pkg/telemetryai/statement_builder_test.go

@@ -0,0 +1,705 @@
+package telemetryai
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetrytraces"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/stretchr/testify/require"
+)
+
+// otelKeysMap seeds the OpenTelemetry gen_ai semantic-convention keys the AI
+// queries reference, so the metadata-backed field resolution succeeds in tests.
+func otelKeysMap() map[string][]*telemetrytypes.TelemetryFieldKey {
+	strKey := func(name string) *telemetrytypes.TelemetryFieldKey {
+		return &telemetrytypes.TelemetryFieldKey{
+			Name:          name,
+			Signal:        telemetrytypes.SignalTraces,
+			FieldContext:  telemetrytypes.FieldContextAttribute,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+		}
+	}
+	numKey := func(name string) *telemetrytypes.TelemetryFieldKey {
+		return &telemetrytypes.TelemetryFieldKey{
+			Name:          name,
+			Signal:        telemetrytypes.SignalTraces,
+			FieldContext:  telemetrytypes.FieldContextAttribute,
+			FieldDataType: telemetrytypes.FieldDataTypeFloat64,
+		}
+	}
+
+	m := make(map[string][]*telemetrytypes.TelemetryFieldKey)
+
+	// gen_ai semconv keys sourced from the single source of truth, mirroring what the
+	// production metadata store surfaces via enrichWithGenAIKeys.
+	for name, def := range telemetrytypes.GenAIFieldDefinitions {
+		keyCopy := def
+		m[name] = []*telemetrytypes.TelemetryFieldKey{&keyCopy}
+	}
+
+	// Extra keys these tests reference that aren't gen_ai semconv definitions.
+	m["gen_ai.user.id"] = []*telemetrytypes.TelemetryFieldKey{strKey("gen_ai.user.id")}
+	m["gen_ai.usage.cost"] = []*telemetrytypes.TelemetryFieldKey{numKey("gen_ai.usage.cost")}
+	m["gen_ai.usage.cached_input_tokens"] = []*telemetrytypes.TelemetryFieldKey{numKey("gen_ai.usage.cached_input_tokens")}
+	m["has_error"] = []*telemetrytypes.TelemetryFieldKey{{
+		Name:          "has_error",
+		Signal:        telemetrytypes.SignalTraces,
+		FieldContext:  telemetrytypes.FieldContextSpan,
+		FieldDataType: telemetrytypes.FieldDataTypeBool,
+	}}
+	return m
+}
+
+// standard test window (ms), matching the traces builder tests.
+const (
+	testStartMs = uint64(1747947419000)
+	testEndMs   = uint64(1747983448000)
+)
+
+func newTestBuilder(t *testing.T) *scopedTraceStatementBuilder {
+	return newTestBuilderWithKeys(t, otelKeysMap())
+}
+
+// newTestBuilderWithKeys mirrors the production wiring in signozquerier's provider.
+// The gen_ai keys are seeded via keysMap here; in production the metadata store
+// surfaces them itself (enrichWithGenAIKeys).
+func newTestBuilderWithKeys(t *testing.T, keysMap map[string][]*telemetrytypes.TelemetryFieldKey) *scopedTraceStatementBuilder {
+	t.Helper()
+	settings := instrumentationtest.New().ToProviderSettings()
+	fm := telemetrytraces.NewFieldMapper()
+	cb := telemetrytraces.NewConditionBuilder(fm)
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore.KeysMap = keysMap
+	fl := flaggertest.New(t)
+	baseCond := NewGenAIBaseConditionProvider()
+	// In production the metadata store enriches gen_ai keys (enrichWithGenAIKeys);
+	// here the mock is seeded directly via keysMap.
+	metadataStore := telemetrytypes.MetadataStore(mockMetadataStore)
+	rewriter := querybuilder.NewAggExprRewriter(settings, nil, fm, cb, nil, fl)
+	traceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
+		settings,
+		metadataStore,
+		fm,
+		cb,
+		rewriter,
+		nil,
+		fl,
+		false,
+		100000,
+	)
+	return NewAITraceStatementBuilder(
+		settings,
+		metadataStore,
+		fm,
+		cb,
+		baseCond,
+		traceStmtBuilder,
+	)
+}
+
+// ---------------------------------------------------------------------------
+// Full-query golden tests
+//
+// Each pins the WHOLE generated statement, with bound args inlined into the `?`
+// placeholders, as ONE self-contained literal — so a failure diff shows the entire
+// query and the expected SQL can be copied straight into a ClickHouse client. The
+// `want` strings are formatted for readability; the comparison is whitespace- and
+// backtick-insensitive (see normalizeSQL), so only the SQL tokens themselves matter.
+//
+// The four trace-list goldens cover the corners of how `matched` is assembled —
+// {no span filter, span filter} × {no aggregate filter, aggregate filter} — plus a
+// mixed filter + multi-key order, plus the delegated span list. Note `matched` selects
+// only the aggregates ORDER BY / HAVING reference; the rest appear only in enrichment.
+//
+// Run `go test ./pkg/telemetryai/ -run TestBuild_FullSQL -v` to also print each query.
+// ---------------------------------------------------------------------------
+
+// renderSQL substitutes bound args into the `?` placeholders so the whole statement
+// reads as one literal SQL string.
+func renderSQL(t *testing.T, stmt *qbtypes.Statement) string {
+	t.Helper()
+	var b strings.Builder
+	argi := 0
+	for i := 0; i < len(stmt.Query); i++ {
+		if stmt.Query[i] == '?' {
+			require.Less(t, argi, len(stmt.Args), "more ? than args in query")
+			b.WriteString(formatArg(stmt.Args[argi]))
+			argi++
+			continue
+		}
+		b.WriteByte(stmt.Query[i])
+	}
+	require.Equal(t, len(stmt.Args), argi, "arg count does not match number of placeholders")
+	return b.String()
+}
+
+func formatArg(a any) string {
+	if s, ok := a.(string); ok {
+		return "'" + s + "'"
+	}
+	return fmt.Sprintf("%v", a)
+}
+
+// normalizeSQL makes the comparison insensitive to formatting: it drops identifier
+// backticks, collapses whitespace runs to a single space, and removes spaces directly
+// inside parentheses. This lets the golden strings be freely indented/wrapped (and
+// written as Go raw literals, which cannot contain backticks) — only the SQL tokens
+// and their order matter.
+func normalizeSQL(s string) string {
+	s = strings.Join(strings.Fields(strings.ReplaceAll(s, "`", "")), " ")
+	s = strings.ReplaceAll(s, "( ", "(")
+	s = strings.ReplaceAll(s, " )", ")")
+	return s
+}
+
+func requireSQLEqual(t *testing.T, want string, stmt *qbtypes.Statement) {
+	t.Helper()
+	got := renderSQL(t, stmt)
+	t.Logf("\n%s", got)
+	require.Equal(t, normalizeSQL(want), normalizeSQL(got))
+}
+
+// No filter: matched selects only the default order key (last_activity_time), WHERE is
+// just window + gate mask, no HAVING.
+func TestBuild_FullSQL_TraceList_NoFilter(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI, Limit: 20,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+WITH matched AS (
+    SELECT trace_id,
+        maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
+    GROUP BY trace_id
+    ORDER BY last_activity_time DESC, trace_id DESC
+    LIMIT 20
+),
+ranked AS (
+    SELECT trace_id, min(start) AS t_start, max(end) AS t_end
+    FROM signoz_traces.distributed_trace_summary
+    WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
+      AND end >= fromUnixTimestamp64Nano(1747947419000000000)
+      AND start < fromUnixTimestamp64Nano(1747983448000000000)
+    GROUP BY trace_id
+),
+buckets AS (
+    SELECT DISTINCT b AS ts_bucket
+    FROM ranked
+    ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
+)
+SELECT trace_id,
+    min(timestamp) AS start_time,
+    max(timestamp) AS end_time,
+    (max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
+    count() AS span_count,
+    anyIf(name, parent_span_id = '') AS root_span_name,
+    any(resource_string_service$$name) AS service.name,
+    countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+    maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
+GROUP BY trace_id
+ORDER BY last_activity_time DESC, trace_id DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Span-level AND trace-level filter, order by the aggregate, pagination. matched selects
+// only output_tokens (the sole aggregate referenced by both ORDER BY and HAVING) — not
+// input_tokens/llm_call_count/last_activity_time. The span predicate widens the WHERE
+// prune and becomes a countIf(...) > 0 existence check alongside the gate countIf.
+func TestBuild_FullSQL_TraceList_SpanAndTraceFilter(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND output_tokens > 1000"},
+			Order:  []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "output_tokens"}}, Direction: qbtypes.OrderDirectionDesc}},
+			Limit:  10, Offset: 30,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+WITH matched AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
+        OR (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true))
+    GROUP BY trace_id
+    HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
+        AND countIf((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
+        AND output_tokens > 1000
+    ORDER BY output_tokens DESC, trace_id DESC
+    LIMIT 10 OFFSET 30
+),
+ranked AS (
+    SELECT trace_id, min(start) AS t_start, max(end) AS t_end
+    FROM signoz_traces.distributed_trace_summary
+    WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
+      AND end >= fromUnixTimestamp64Nano(1747947419000000000)
+      AND start < fromUnixTimestamp64Nano(1747983448000000000)
+    GROUP BY trace_id
+),
+buckets AS (
+    SELECT DISTINCT b AS ts_bucket
+    FROM ranked
+    ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
+)
+SELECT trace_id,
+    min(timestamp) AS start_time,
+    max(timestamp) AS end_time,
+    (max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
+    count() AS span_count,
+    anyIf(name, parent_span_id = '') AS root_span_name,
+    any(resource_string_service$$name) AS service.name,
+    countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+    maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
+GROUP BY trace_id
+ORDER BY output_tokens DESC, trace_id DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Aggregate-only filter (no span filter). WHERE prune is NOT widened, there is no
+// gate/span countIf, just the aggregate HAVING. `trace.output_tokens` rewrites to the
+// output_tokens alias. matched selects output_tokens (HAVING) + last_activity_time (default order).
+func TestBuild_FullSQL_TraceList_AggregateFilterOnly(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
+			Limit:  20,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+WITH matched AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+        maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
+    GROUP BY trace_id
+    HAVING output_tokens > 1000
+    ORDER BY last_activity_time DESC, trace_id DESC
+    LIMIT 20
+),
+ranked AS (
+    SELECT trace_id, min(start) AS t_start, max(end) AS t_end
+    FROM signoz_traces.distributed_trace_summary
+    WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
+      AND end >= fromUnixTimestamp64Nano(1747947419000000000)
+      AND start < fromUnixTimestamp64Nano(1747983448000000000)
+    GROUP BY trace_id
+),
+buckets AS (
+    SELECT DISTINCT b AS ts_bucket
+    FROM ranked
+    ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
+)
+SELECT trace_id,
+    min(timestamp) AS start_time,
+    max(timestamp) AS end_time,
+    (max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
+    count() AS span_count,
+    anyIf(name, parent_span_id = '') AS root_span_name,
+    any(resource_string_service$$name) AS service.name,
+    countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+    maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
+GROUP BY trace_id
+ORDER BY last_activity_time DESC, trace_id DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Span-only filter (no aggregate filter). WHERE is widened; HAVING has the gate + span
+// countIf pair but no trailing aggregate. `has_error = true` resolves to a
+// materialized-column predicate (not a map access). matched selects only the default order key.
+func TestBuild_FullSQL_TraceList_SpanFilterOnly(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "has_error = true"},
+			Limit:  20,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+WITH matched AS (
+    SELECT trace_id,
+        maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
+        OR has_error = true)
+    GROUP BY trace_id
+    HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
+        AND countIf(has_error = true) > 0
+    ORDER BY last_activity_time DESC, trace_id DESC
+    LIMIT 20
+),
+ranked AS (
+    SELECT trace_id, min(start) AS t_start, max(end) AS t_end
+    FROM signoz_traces.distributed_trace_summary
+    WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
+      AND end >= fromUnixTimestamp64Nano(1747947419000000000)
+      AND start < fromUnixTimestamp64Nano(1747983448000000000)
+    GROUP BY trace_id
+),
+buckets AS (
+    SELECT DISTINCT b AS ts_bucket
+    FROM ranked
+    ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
+)
+SELECT trace_id,
+    min(timestamp) AS start_time,
+    max(timestamp) AS end_time,
+    (max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
+    count() AS span_count,
+    anyIf(name, parent_span_id = '') AS root_span_name,
+    any(resource_string_service$$name) AS service.name,
+    countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+    maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
+GROUP BY trace_id
+ORDER BY last_activity_time DESC, trace_id DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Mixed filter (two span predicates AND'd into one existence check + an aggregate) with
+// a two-key order on different aggregates than the filter. matched selects input_tokens
+// + last_activity_time (ORDER BY) and output_tokens (HAVING) — three of four; llm_call_count is not.
+func TestBuild_FullSQL_TraceList_MixedFiltersMultiOrder(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o' AND has_error = true AND output_tokens > 500"},
+			Order: []qbtypes.OrderBy{
+				{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "input_tokens"}}, Direction: qbtypes.OrderDirectionDesc},
+				{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "last_activity_time"}}, Direction: qbtypes.OrderDirectionAsc},
+			},
+			Limit: 15,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+WITH matched AS (
+    SELECT trace_id,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+        sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+        maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+    FROM signoz_traces.distributed_signoz_index_v3
+    WHERE timestamp >= '1747947419000000000'
+      AND timestamp < '1747983448000000000'
+      AND ts_bucket_start >= 1747945619
+      AND ts_bucket_start <= 1747983448
+      AND ((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
+        OR ((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true))
+    GROUP BY trace_id
+    HAVING countIf((mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) > 0
+        AND countIf(((attributes_string['gen_ai.request.model'] = 'gpt-4o' AND mapContains(attributes_string, 'gen_ai.request.model') = true) AND has_error = true)) > 0
+        AND output_tokens > 500
+    ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
+    LIMIT 15
+),
+ranked AS (
+    SELECT trace_id, min(start) AS t_start, max(end) AS t_end
+    FROM signoz_traces.distributed_trace_summary
+    WHERE trace_id GLOBAL IN (SELECT trace_id FROM matched)
+      AND end >= fromUnixTimestamp64Nano(1747947419000000000)
+      AND start < fromUnixTimestamp64Nano(1747983448000000000)
+    GROUP BY trace_id
+),
+buckets AS (
+    SELECT DISTINCT b AS ts_bucket
+    FROM ranked
+    ARRAY JOIN range(toUInt64(intDiv(toUnixTimestamp(t_start), 1800) * 1800 - 1800), toUInt64(intDiv(toUnixTimestamp(t_end), 1800) * 1800 + 1800), 1800) AS b
+)
+SELECT trace_id,
+    min(timestamp) AS start_time,
+    max(timestamp) AS end_time,
+    (max(toUnixTimestamp64Nano(timestamp) + duration_nano) - min(toUnixTimestamp64Nano(timestamp))) AS duration_nano,
+    count() AS span_count,
+    anyIf(name, parent_span_id = '') AS root_span_name,
+    any(resource_string_service$$name) AS service.name,
+    countIf(mapContains(attributes_string, 'gen_ai.request.model') = true) AS llm_call_count,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)) AS input_tokens,
+    sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens,
+    maxIf(timestamp, (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)) AS last_activity_time
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE ts_bucket_start GLOBAL IN (SELECT ts_bucket FROM buckets)
+  AND trace_id GLOBAL IN (SELECT trace_id FROM ranked)
+GROUP BY trace_id
+ORDER BY input_tokens DESC, last_activity_time ASC, trace_id DESC
+SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
+`, stmt)
+}
+
+// Span list (requestType raw): delegated to the traces builder with the gate ANDed
+// into the user filter, so only gen_ai spans matching the filter come back. Standard
+// span columns, single SELECT (no CTE pipeline).
+func TestBuild_FullSQL_SpanList_Raw(t *testing.T) {
+	b := newTestBuilder(t)
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeRaw,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
+			Limit:  10,
+		}, nil)
+	require.NoError(t, err)
+
+	requireSQLEqual(t, `
+SELECT timestamp AS timestamp, trace_id AS trace_id, span_id AS span_id,
+    trace_state AS trace_state, parent_span_id AS parent_span_id, flags AS flags,
+    name AS name, kind AS kind, kind_string AS kind_string, duration_nano AS duration_nano,
+    status_code AS status_code, status_message AS status_message,
+    status_code_string AS status_code_string, events AS events, links AS links,
+    response_status_code AS response_status_code, external_http_url AS external_http_url,
+    http_url AS http_url, external_http_method AS external_http_method,
+    http_method AS http_method, http_host AS http_host, db_name AS db_name,
+    db_operation AS db_operation, has_error AS has_error, is_remote AS is_remote,
+    attributes_string, attributes_number, attributes_bool, resources_string
+FROM signoz_traces.distributed_signoz_index_v3
+WHERE (((mapContains(attributes_string, 'gen_ai.request.model') = true
+        OR mapContains(attributes_string, 'gen_ai.tool.name') = true
+        OR mapContains(attributes_string, 'gen_ai.agent.name') = true))
+    AND ((attributes_string['gen_ai.request.model'] = 'gpt-4o-mini'
+        AND mapContains(attributes_string, 'gen_ai.request.model') = true)))
+  AND timestamp >= '1747947419000000000'
+  AND timestamp < '1747983448000000000'
+  AND ts_bucket_start >= 1747945619
+  AND ts_bucket_start <= 1747983448
+LIMIT 10
+`, stmt)
+}
+
+// ---------------------------------------------------------------------------
+// Behavior / branch tests not covered by the goldens above
+// ---------------------------------------------------------------------------
+
+// Trace-level and span-level predicates may not be OR-combined.
+func TestBuild_TraceList_TraceOrSpanMixRejected(t *testing.T) {
+	b := newTestBuilder(t)
+	query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal: telemetrytypes.SignalTraces,
+		Source: telemetrytypes.SourceAI,
+		Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000 OR gen_ai.request.model = 'x'"},
+		Limit:  10,
+	}
+	_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "cannot be combined")
+}
+
+// An output-only aggregate (span_count / duration_nano) can be displayed but not used
+// in the aggregate filter or ORDER BY — it is not computable in the matched pass.
+func TestBuild_TraceList_OutputOnlyAggregateRejected(t *testing.T) {
+	b := newTestBuilder(t)
+
+	// filter by span_count -> rejected
+	_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Filter: &qbtypes.Filter{Expression: "span_count > 3"},
+		}, nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "span_count")
+
+	// order by duration_nano -> rejected
+	_, err = b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+			Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}}, Direction: qbtypes.OrderDirectionDesc}},
+		}, nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unsupported order key")
+}
+
+// A HAVING referencing a non-aggregate column is rejected.
+func TestBuild_TraceList_Having_UnknownColumn(t *testing.T) {
+	b := newTestBuilder(t)
+	query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal: telemetrytypes.SignalTraces,
+		Source: telemetrytypes.SourceAI,
+		Having: &qbtypes.Having{Expression: "service.name > 1"}, // not an aggregate column
+		Limit:  10,
+	}
+	_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
+	require.Error(t, err)
+}
+
+// Ordering by an unknown key is rejected.
+func TestBuild_TraceList_UnsupportedOrderKey(t *testing.T) {
+	b := newTestBuilder(t)
+	query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal: telemetrytypes.SignalTraces,
+		Source: telemetrytypes.SourceAI,
+		Order: []qbtypes.OrderBy{
+			{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "http.request.method"}}, Direction: qbtypes.OrderDirectionDesc},
+		},
+	}
+	_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unsupported order key")
+}
+
+// With no limit set, the builder applies the default of 100.
+func TestBuild_TraceList_DefaultLimit(t *testing.T) {
+	b := newTestBuilder(t)
+	query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal: telemetrytypes.SignalTraces,
+		Source: telemetrytypes.SourceAI,
+	}
+	stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace, query, nil)
+	require.NoError(t, err)
+	require.Contains(t, stmt.Query, "LIMIT ?")
+	require.Contains(t, stmt.Args, 100)
+}
+
+// Only trace list and span list (raw) are supported; distribution is not.
+func TestBuild_UnsupportedRequestType(t *testing.T) {
+	b := newTestBuilder(t)
+	query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+		Signal: telemetrytypes.SignalTraces,
+		Source: telemetrytypes.SourceAI,
+		Aggregations: []qbtypes.TraceAggregation{
+			{Expression: "count()"},
+		},
+	}
+	_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeDistribution, query, nil)
+	require.ErrorIs(t, err, ErrUnsupportedRequestType)
+}
+
+// ---------------------------------------------------------------------------
+// Filter operator resolution
+//
+// The goldens pin the CTE structure; these pin how each filter OPERATOR resolves into
+// SQL (the part that varies with the operator, not the pipeline). Span-level operators
+// resolve to a predicate that appears both in the widened WHERE prune and as a
+// countIf(...) > 0 existence check; aggregate operators become a HAVING (values inlined).
+// ---------------------------------------------------------------------------
+
+func TestBuild_TraceList_SpanFilterOperatorResolution(t *testing.T) {
+	b := newTestBuilder(t)
+	build := func(t *testing.T, expr string) string {
+		stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+			qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+				Filter: &qbtypes.Filter{Expression: expr}, Limit: 5,
+			}, nil)
+		require.NoError(t, err)
+		return stmt.Query
+	}
+
+	cases := []struct{ name, expr, frag string }{
+		{"not_equal", "gen_ai.request.model != 'gpt-4o'",
+			"attributes_string['gen_ai.request.model'] <> ?"},
+		{"in", "gen_ai.request.model IN ('gpt-4o', 'gpt-4')",
+			"(attributes_string['gen_ai.request.model'] = ? OR attributes_string['gen_ai.request.model'] = ?) AND mapContains(attributes_string, 'gen_ai.request.model') = ?"},
+		{"exists", "gen_ai.user.id EXISTS", // non-gate key, so the fragment is unambiguous
+			"mapContains(attributes_string, 'gen_ai.user.id') = ?"},
+		{"not_exists", "gen_ai.user.id NOT EXISTS",
+			"mapContains(attributes_string, 'gen_ai.user.id') <> ?"},
+		{"contains", "gen_ai.request.model CONTAINS 'gpt'", // case-insensitive
+			"LOWER(attributes_string['gen_ai.request.model']) LIKE LOWER(?)"},
+		{"like", "gen_ai.request.model LIKE 'gpt%'",
+			"attributes_string['gen_ai.request.model'] LIKE ?"},
+		{"numeric_gte", "gen_ai.usage.output_tokens >= 100", // span attr (Float64), distinct from the output_tokens aggregate
+			"toFloat64(attributes_number['gen_ai.usage.output_tokens']) >= ? AND mapContains(attributes_number, 'gen_ai.usage.output_tokens') = ?"},
+		{"not_group", "NOT (gen_ai.request.model = 'gpt-4o')",
+			"NOT (((attributes_string['gen_ai.request.model'] = ? AND mapContains(attributes_string, 'gen_ai.request.model') = ?)))"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			// the resolved predicate appears in the widened WHERE prune and (wrapped) in
+			// the countIf existence check; the goldens pin the countIf structure, here we
+			// pin only the operator's resolution.
+			require.Contains(t, build(t, c.expr), c.frag)
+		})
+	}
+}
+
+func TestBuild_TraceList_AggregateFilterOperatorResolution(t *testing.T) {
+	b := newTestBuilder(t)
+	build := func(t *testing.T, expr string) (string, error) {
+		stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTrace,
+			qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+				Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
+				Filter: &qbtypes.Filter{Expression: expr}, Limit: 5,
+			}, nil)
+		if err != nil {
+			return "", err
+		}
+		return stmt.Query, nil
+	}
+
+	// values are inlined by the HAVING rewriter (not parameterized).
+	cases := []struct{ name, expr, having string }{
+		{"less_than", "output_tokens < 500", "HAVING output_tokens < 500"},
+		{"not_equal", "output_tokens != 0", "HAVING output_tokens != 0"},
+		{"range_and", "output_tokens >= 500 AND output_tokens <= 1000", "HAVING (output_tokens >= 500 AND output_tokens <= 1000)"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			q, err := build(t, c.expr)
+			require.NoError(t, err)
+			require.Contains(t, q, c.having)
+		})
+	}
+
+	// BETWEEN is not supported by the HAVING rewriter — surfaced as an error, not silently wrong.
+	t.Run("between_unsupported", func(t *testing.T) {
+		_, err := build(t, "output_tokens BETWEEN 500 AND 1000")
+		require.Error(t, err)
+	})
+}

pkg/telemetrymetadata/metadata.go

@@ -1190,6 +1190,27 @@ func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFi
 	return keys
 }
 
+// enrichWithGenAIKeys adds keys that can be queried for GenAI signals, even though they have not been ingested yet.
+func enrichWithGenAIKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
+	for _, selector := range selectors {
+		if selector.Signal != telemetrytypes.SignalTraces && selector.Signal != telemetrytypes.SignalUnspecified {
+			continue
+		}
+		for name, def := range telemetrytypes.GenAIFieldDefinitions {
+			if len(keys[name]) > 0 {
+				continue // already resolved from ingested data
+			}
+			if !selectorMatchesIntrinsicField(selector, def) {
+				continue
+			}
+			keyCopy := def
+			keys[name] = []*telemetrytypes.TelemetryFieldKey{&keyCopy}
+		}
+	}
+
+	return keys
+}
+
 func selectorMatchesIntrinsicField(selector *telemetrytypes.FieldKeySelector, definition telemetrytypes.TelemetryFieldKey) bool {
 	if selector.FieldContext != telemetrytypes.FieldContextUnspecified && selector.FieldContext != definition.FieldContext {
 		return false
@@ -1275,6 +1296,7 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
 
 	applyBackwardCompatibleKeys(mapOfKeys)
 	mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, selectors)
+	mapOfKeys = enrichWithGenAIKeys(mapOfKeys, selectors)
 
 	return mapOfKeys, complete, nil
 }
@@ -1353,6 +1375,7 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 
 	applyBackwardCompatibleKeys(mapOfKeys)
 	mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, fieldKeySelectors)
+	mapOfKeys = enrichWithGenAIKeys(mapOfKeys, fieldKeySelectors)
 
 	return mapOfKeys, complete, nil
 }

pkg/types/llmpricingruletypes/pricing.go

@@ -16,13 +16,6 @@ import (
 const (
 	LLMCostFeatureType agentConf.AgentFeatureType = "llm_pricing"
 
-	GenAIRequestModel                  = "gen_ai.request.model"
-	GenAIProviderName                  = "gen_ai.provider.name"
-	GenAIUsageInputTokens              = "gen_ai.usage.input_tokens"
-	GenAIUsageOutputTokens             = "gen_ai.usage.output_tokens"
-	GenAIUsageCacheReadInputTokens     = "gen_ai.usage.cache_read.input_tokens"
-	GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
-
 	SignozGenAICostInput      = "_signoz.gen_ai.cost_input"
 	SignozGenAICostOutput     = "_signoz.gen_ai.cost_output"
 	SignozGenAICostCacheRead  = "_signoz.gen_ai.cost_cache_read"

pkg/types/llmpricingruletypes/processorconfig.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 
 	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"gopkg.in/yaml.v3"
 )
 
@@ -83,11 +84,11 @@ func buildProcessorConfig(rules []*LLMPricingRule) *LLMPricingRuleProcessorConfi
 
 	return &LLMPricingRuleProcessorConfig{
 		Attrs: LLMPricingRuleProcessorAttrs{
-			Model:      GenAIRequestModel,
-			In:         GenAIUsageInputTokens,
-			Out:        GenAIUsageOutputTokens,
-			CacheRead:  GenAIUsageCacheReadInputTokens,
-			CacheWrite: GenAIUsageCacheCreationInputTokens,
+			Model:      telemetrytypes.GenAIRequestModel,
+			In:         telemetrytypes.GenAIUsageInputTokens,
+			Out:        telemetrytypes.GenAIUsageOutputTokens,
+			CacheRead:  telemetrytypes.GenAIUsageCacheReadInputTokens,
+			CacheWrite: telemetrytypes.GenAIUsageCacheCreationInputTokens,
 		},
 		DefaultPricing: LLMPricingRuleProcessorDefaultPricing{
 			Rules: pricingRules,

pkg/types/telemetrytypes/genai_semconv.go

@@ -0,0 +1,32 @@
+package telemetrytypes
+
+// OpenTelemetry gen_ai semantic-convention attribute keys. Single source of truth
+// shared by the AI query builder and the LLM pricing pipeline.
+const (
+	GenAIRequestModel = "gen_ai.request.model"
+	GenAIToolName     = "gen_ai.tool.name"
+	GenAIAgentName    = "gen_ai.agent.name"
+	GenAIProviderName = "gen_ai.provider.name"
+
+	GenAIUsageInputTokens              = "gen_ai.usage.input_tokens"
+	GenAIUsageOutputTokens             = "gen_ai.usage.output_tokens"
+	GenAIUsageCacheReadInputTokens     = "gen_ai.usage.cache_read.input_tokens"
+	GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
+)
+
+// GenAIFieldDefinitions are the gen_ai semantic-convention span attributes the AI
+// query builder relies on. They are surfaced by the metadata store for trace
+// queries regardless of whether they have been ingested yet, so the AI gate/columns
+// resolve on a fresh install (mirrors intrinsic metric keys). String keys are the
+// gate; the usage keys are numeric.
+var GenAIFieldDefinitions = map[string]TelemetryFieldKey{
+	GenAIRequestModel: {Name: GenAIRequestModel, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
+	GenAIToolName:     {Name: GenAIToolName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
+	GenAIAgentName:    {Name: GenAIAgentName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
+	GenAIProviderName: {Name: GenAIProviderName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
+
+	GenAIUsageInputTokens:              {Name: GenAIUsageInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
+	GenAIUsageOutputTokens:             {Name: GenAIUsageOutputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
+	GenAIUsageCacheReadInputTokens:     {Name: GenAIUsageCacheReadInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
+	GenAIUsageCacheCreationInputTokens: {Name: GenAIUsageCacheCreationInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
+}

pkg/types/telemetrytypes/source.go

@@ -9,6 +9,7 @@ type Source struct {
 var (
 	SourceAudit       = Source{valuer.NewString("audit")}
 	SourceMeter       = Source{valuer.NewString("meter")}
+	SourceAI          = Source{valuer.NewString("ai")}
 	SourceUnspecified = Source{valuer.NewString("")}
 )
 
@@ -17,5 +18,6 @@ var (
 func (Source) Enum() []any {
 	return []any{
 		SourceMeter,
+		SourceAI,
 	}
 }

tests/fixtures/querier.py

@@ -79,10 +79,13 @@ class BuilderQuery:
     name: str = "A"
     source: str | None = None
     limit: int | None = None
+    offset: int | None = None
     filter_expression: str | None = None
+    having_expression: str | None = None
     select_fields: list[TelemetryFieldKey] | None = None
     order: list[OrderBy] | None = None
     aggregations: list[Aggregation | MetricAggregation] | None = None
+    group_by: list[TelemetryFieldKey] | None = None
     step_interval: int | None = None
 
     def to_dict(self) -> dict:
@@ -94,14 +97,20 @@ class BuilderQuery:
             spec["source"] = self.source
         if self.limit is not None:
             spec["limit"] = self.limit
+        if self.offset is not None:
+            spec["offset"] = self.offset
         if self.filter_expression:
             spec["filter"] = {"expression": self.filter_expression}
+        if self.having_expression:
+            spec["having"] = {"expression": self.having_expression}
         if self.select_fields:
             spec["selectFields"] = [f.to_dict() for f in self.select_fields]
         if self.order:
             spec["order"] = [o.to_dict() if hasattr(o, "to_dict") else o for o in self.order]
         if self.aggregations:
             spec["aggregations"] = [agg.to_dict() if hasattr(agg, "to_dict") else agg for agg in self.aggregations]
+        if self.group_by:
+            spec["groupBy"] = [k.to_dict() for k in self.group_by]
         if self.step_interval is not None:
             spec["stepInterval"] = self.step_interval
 

tests/integration/tests/querierai/01_ai_traces.py

@@ -0,0 +1,544 @@
+"""
+Integration tests for source="ai" over the traces signal.
+
+These ingest OpenTelemetry gen_ai spans into real ClickHouse via the insert_traces
+fixture and exercise the actual /api/v5/query_range API, so they validate the whole
+path: payload -> AI statement builder -> ClickHouse -> response.
+
+Data shape (generic OTel gen_ai semantic conventions):
+  - a root span (no gen_ai attributes)
+  - an LLM span carrying gen_ai.request.model (str) and numeric usage attributes
+    (gen_ai.usage.input_tokens / output_tokens / cost) plus gen_ai.user.id
+Each test tags its spans with a unique service.name and filters on it, so tests do
+not interfere with each other's data.
+"""
+
+import json
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    BuilderQuery,
+    MetricAggregation,
+    OrderBy,
+    RequestType,
+    TelemetryFieldKey,
+    make_query_request,
+)
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+
+def _ai_trace(
+    *,
+    now: datetime,
+    service: str,
+    user: str,
+    in_tokens: int,
+    out_tokens: int,
+    cost: float,
+    llm_duration_s: float = 1.0,
+    error: bool = False,
+) -> list[Traces]:
+    """A minimal AI trace: root span + one LLM span with gen_ai attributes."""
+    trace_id = TraceIdGenerator.trace_id()
+    root_id = TraceIdGenerator.span_id()
+    llm_id = TraceIdGenerator.span_id()
+    resources = {"service.name": service, "deployment.environment": "production"}
+
+    root = Traces(
+        timestamp=now - timedelta(seconds=5),
+        duration=timedelta(seconds=llm_duration_s + 0.1),
+        trace_id=trace_id,
+        span_id=root_id,
+        parent_span_id="",
+        name="POST /api/chat",
+        kind=TracesKind.SPAN_KIND_SERVER,
+        status_code=TracesStatusCode.STATUS_CODE_OK,
+        resources=resources,
+        attributes={"http.request.method": "POST"},
+    )
+    llm = Traces(
+        timestamp=now - timedelta(seconds=4),
+        duration=timedelta(seconds=llm_duration_s),
+        trace_id=trace_id,
+        span_id=llm_id,
+        parent_span_id=root_id,
+        name="chat gpt-4o-mini",
+        kind=TracesKind.SPAN_KIND_CLIENT,
+        status_code=(TracesStatusCode.STATUS_CODE_ERROR if error else TracesStatusCode.STATUS_CODE_OK),
+        resources=resources,
+        attributes={
+            "gen_ai.request.model": "gpt-4o-mini",
+            "gen_ai.system": "openai",
+            "gen_ai.user.id": user,
+            # numeric values land in attributes_number
+            "gen_ai.usage.input_tokens": in_tokens,
+            "gen_ai.usage.output_tokens": out_tokens,
+            "gen_ai.usage.cost": cost,
+        },
+    )
+    return [root, llm]
+
+
+def _non_ai_trace(*, now: datetime, service: str) -> list[Traces]:
+    """A plain HTTP trace with no gen_ai attributes; must be excluded by the AI gate."""
+    trace_id = TraceIdGenerator.trace_id()
+    span_id = TraceIdGenerator.span_id()
+    return [
+        Traces(
+            timestamp=now - timedelta(seconds=4),
+            duration=timedelta(seconds=1),
+            trace_id=trace_id,
+            span_id=span_id,
+            parent_span_id="",
+            name="GET /health",
+            kind=TracesKind.SPAN_KIND_SERVER,
+            status_code=TracesStatusCode.STATUS_CODE_OK,
+            resources={"service.name": service},
+            attributes={"http.request.method": "GET"},
+        )
+    ]
+
+
+def _window_ms(now: datetime) -> tuple[int, int]:
+    start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
+    end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
+    return start_ms, end_ms
+
+
+def test_ai_list_excludes_non_ai(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Trace-list panel (requestType="trace"): returns AI traces and excludes the
+    non-AI trace. Asserts on the raw response payload to stay agnostic to the exact
+    row schema.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-list"
+
+    ai = _ai_trace(now=now, service=service, user="alice", in_tokens=100, out_tokens=20, cost=0.5)
+    non_ai = _non_ai_trace(now=now, service=service)
+    ai_trace_id = ai[0].trace_id
+    non_ai_trace_id = non_ai[0].trace_id
+    insert_traces(ai + non_ai)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        limit=10,
+    )
+
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    body = json.dumps(response.json())
+    assert ai_trace_id in body, f"expected AI trace {ai_trace_id} in list response"
+    assert non_ai_trace_id not in body, f"non-AI trace {non_ai_trace_id} should be excluded by the gate"
+
+
+def _ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Traces]:
+    """
+    Root + one LLM span + one tool span + one agent span. The gate matches all three
+    child spans, but only the LLM span carries gen_ai.request.model.
+    """
+    trace_id = TraceIdGenerator.trace_id()
+    root_id = TraceIdGenerator.span_id()
+    resources = {"service.name": service, "deployment.environment": "production"}
+
+    def _span(name, kind, attributes, offset_s):
+        return Traces(
+            timestamp=now - timedelta(seconds=offset_s),
+            duration=timedelta(seconds=0.5),
+            trace_id=trace_id,
+            span_id=TraceIdGenerator.span_id(),
+            parent_span_id=root_id,
+            name=name,
+            kind=kind,
+            status_code=TracesStatusCode.STATUS_CODE_OK,
+            resources=resources,
+            attributes=attributes,
+        )
+
+    root = Traces(
+        timestamp=now - timedelta(seconds=5),
+        duration=timedelta(seconds=4),
+        trace_id=trace_id,
+        span_id=root_id,
+        parent_span_id="",
+        name="POST /api/chat",
+        kind=TracesKind.SPAN_KIND_SERVER,
+        status_code=TracesStatusCode.STATUS_CODE_OK,
+        resources=resources,
+        attributes={"http.request.method": "POST"},
+    )
+    llm = _span("chat gpt-4o-mini", TracesKind.SPAN_KIND_CLIENT, {
+        "gen_ai.request.model": "gpt-4o-mini",
+        "gen_ai.system": "openai",
+        "gen_ai.user.id": user,
+        "gen_ai.usage.input_tokens": 100,
+        "gen_ai.usage.output_tokens": 20,
+    }, 4)
+    tool = _span("execute_tool", TracesKind.SPAN_KIND_INTERNAL, {
+        "gen_ai.tool.name": "get_weather",
+        "gen_ai.tool.type": "function",
+    }, 3)
+    agent = _span("agent.step", TracesKind.SPAN_KIND_INTERNAL, {
+        "gen_ai.agent.name": "chat-agent",
+    }, 2)
+    return [root, llm, tool, agent]
+
+
+def test_ai_list_llm_call_count_counts_llm_only(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    llm_call_count counts LLM spans only (gen_ai.request.model), not the full gate:
+    a trace with 1 LLM + 1 tool + 1 agent span (4 spans incl. root) reports
+    llm_call_count == 1 and span_count == 4.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-llmcount"
+    insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="alice"))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        limit=10,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    rows = response.json()["data"]["data"]["results"][0]["rows"]
+    assert len(rows) == 1, f"expected exactly one AI trace, got: {rows}"
+    data = rows[0]["data"]
+    assert data["llm_call_count"] == 1, f"llm_call_count should count LLM spans only: {data}"
+    assert data["span_count"] == 4, f"span_count should include all spans: {data}"
+
+
+def test_ai_list_having_aggregate_filter(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Aggregate filter written in the SAME filter box: the span-level predicate narrows
+    to the service, the trace-level `output_tokens > 100` keeps the large-token
+    trace and drops the small one (split internally into WHERE + HAVING).
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-having"
+
+    small = _ai_trace(now=now, service=service, user="alice", in_tokens=10, out_tokens=20, cost=0.1)
+    large = _ai_trace(now=now, service=service, user="bob", in_tokens=10, out_tokens=500, cost=0.2)
+    small_id = small[0].trace_id
+    large_id = large[0].trace_id
+    insert_traces(small + large)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}' AND output_tokens > 100",
+        limit=10,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    body = json.dumps(response.json())
+    assert large_id in body, f"trace with 500 out-tokens should pass output_tokens > 100"
+    assert small_id not in body, f"trace with 20 out-tokens should be filtered out by HAVING"
+
+
+def test_ai_list_order_limit_offset(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Trace list honors order by (aggregate column) + limit + offset (pagination)."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-order"
+
+    traces: list[Traces] = []
+    for out in (100, 200, 300, 400, 500):
+        traces += _ai_trace(now=now, service=service, user="u", in_tokens=10, out_tokens=out, cost=0.1)
+    insert_traces(traces)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    def page(offset: int) -> list[int]:
+        query = BuilderQuery(
+            signal="traces",
+            source="ai",
+            name="A",
+            filter_expression=f"service.name = '{service}'",
+            order=[OrderBy(key=TelemetryFieldKey(name="output_tokens"), direction="desc")],
+            limit=2,
+            offset=offset,
+        )
+        resp = make_query_request(
+            signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+        )
+        assert resp.status_code == HTTPStatus.OK, resp.text
+        rows = resp.json()["data"]["data"]["results"][0]["rows"]
+        return [int(r["data"]["output_tokens"]) for r in rows]
+
+    assert page(0) == [500, 400], "first page: highest output_tokens, desc"
+    assert page(2) == [300, 200], "second page (offset 2): next two, desc"
+
+
+def test_ai_span_list_limit(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """Span list honors limit (delegated raw path): 6 gen_ai spans available, capped to 4."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-spanlimit"
+    insert_traces(
+        _ai_trace_mixed_spans(now=now, service=service, user="a")
+        + _ai_trace_mixed_spans(now=now, service=service, user="b")
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        limit=4,
+    )
+    resp = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
+    )
+    assert resp.status_code == HTTPStatus.OK, resp.text
+    rows = resp.json()["data"]["data"]["results"][0]["rows"]
+    assert len(rows) == 4, f"limit should cap at 4 (6 gen_ai spans available), got {len(rows)}"
+
+
+def test_ai_span_list_excludes_non_gen_ai_spans(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Span list (requestType=raw): returns only the gen_ai spans (LLM/tool/agent); the
+    root span of the same trace (no gen_ai attributes) is excluded by the span-level gate.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-spanlist"
+    insert_traces(_ai_trace_mixed_spans(now=now, service=service, user="alice"))
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}'",
+        select_fields=[TelemetryFieldKey(name="name", field_context="span")],
+        limit=50,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    rows = response.json()["data"]["data"]["results"][0]["rows"]
+    names = sorted(r["data"]["name"] for r in rows)
+    assert names == ["agent.step", "chat gpt-4o-mini", "execute_tool"], names
+    assert "POST /api/chat" not in names  # root span excluded
+
+
+def test_ai_list_having_or_aggregates(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Two aggregate conditions OR-ed within the filter box (regression guard for OR-group
+    whitespace handling): output_tokens > 100 OR span_count > 100 keeps only the
+    large-token trace (span_count is 2 for both, so that branch never matches).
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-having-or"
+
+    small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
+    large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
+    small_id, large_id = small[0].trace_id, large[0].trace_id
+    insert_traces(small + large)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}' AND (output_tokens > 100 OR span_count > 100)",
+        limit=10,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    body = json.dumps(response.json())
+    assert large_id in body
+    assert small_id not in body
+
+
+def test_ai_list_having_trace_context_prefix(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """The `trace.` context prefix on an aggregate column works like the bare name."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    service = "ai-it-trace-ctx"
+
+    small = _ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=20, cost=0.1)
+    large = _ai_trace(now=now, service=service, user="b", in_tokens=10, out_tokens=500, cost=0.2)
+    small_id, large_id = small[0].trace_id, large[0].trace_id
+    insert_traces(small + large)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="traces",
+        source="ai",
+        name="A",
+        filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
+        limit=10,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    body = json.dumps(response.json())
+    assert large_id in body
+    assert small_id not in body
+
+
+def test_ai_source_rejected_on_logs(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    """source=ai is only valid for traces; on logs it must be a validation error."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(signal="logs", source="ai", name="A", limit=5)
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "traces signal" in response.text
+
+
+def test_ai_source_rejected_on_metrics(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    """source=ai on metrics must be a validation error, not a silent normal query."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    query = BuilderQuery(
+        signal="metrics",
+        source="ai",
+        name="A",
+        aggregations=[MetricAggregation(
+            metric_name="system_memory_usage",
+            time_aggregation="avg",
+            space_aggregation="avg",
+            temporality="unspecified",
+        )],
+        step_interval=60,
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "traces signal" in response.text
+
+
+def test_ai_list_rejects_aggregate_or_span_filter(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    """
+    Aggregate (HAVING) columns may not be OR-ed with span-level keys in the trace
+    list; a span-OR-span filter is fine.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms, end_ms = _window_ms(now)
+
+    # aggregate OR span -> rejected
+    bad = BuilderQuery(
+        signal="traces", source="ai", name="A", limit=10,
+        filter_expression="output_tokens > 1000 OR service.name = 'ai-it-orfilter'",
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [bad.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "cannot be combined" in response.text
+
+    # span OR span -> accepted (empty result is fine; just not an error)
+    ok = BuilderQuery(
+        signal="traces", source="ai", name="A", limit=10,
+        filter_expression="service.name = 'ai-it-orfilter' OR has_error = true",
+    )
+    response = make_query_request(
+        signoz, token, start_ms, end_ms, [ok.to_dict()], request_type="trace"
+    )
+    assert response.status_code == HTTPStatus.OK, response.text