fix: filter duckdb usage window before dedup and use shared trend counting

Author: mariusvniekerk

internal/duckdb/analytics_usage.go

@@ -1662,11 +1662,8 @@ func (s *Store) GetTrendsTerms(
 			continue
 		}
 		messageCounts[pos]++
-		lower := strings.ToLower(content)
 		for i, term := range terms {
-			for _, variant := range trendVariants(term) {
-				counts[i][pos] += strings.Count(lower, strings.ToLower(variant))
-			}
+			counts[i][pos] += db.CountTrendOccurrences(content, term)
 		}
 	}
 	if err := rows.Err(); err != nil {
@@ -1677,19 +1674,6 @@ func (s *Store) GetTrendsTerms(
 	), nil
 }
 
-func trendVariants(term db.TrendTermInput) []string {
-	if len(term.Matchers) > 0 {
-		return term.Matchers
-	}
-	if len(term.Variants) > 0 {
-		return term.Variants
-	}
-	if term.Term != "" {
-		return []string{term.Term}
-	}
-	return nil
-}
-
 type duckRates struct {
 	input         float64
 	output        float64
@@ -1946,6 +1930,20 @@ func duckUsageLocalDateSQL(f db.UsageFilter) (string, any) {
 func duckUsageCTE(f db.UsageFilter, sessionID string) (string, []any) {
 	rawSQL, args := duckUsageRawSQL(f, sessionID)
 	localDateSQL, localDateArg := duckUsageLocalDateSQL(f)
+	// Apply the local-date window BEFORE deduping so an out-of-range
+	// duplicate (pulled in by the padded UTC bounds) cannot win
+	// dedup_rank = 1 and suppress the in-range row. Mirrors the
+	// dedup-after-date-filter order in internal/db/usage.go.
+	datePred := "TRUE"
+	var dateArgs []any
+	if f.From != "" {
+		datePred += " AND local_date >= ?"
+		dateArgs = append(dateArgs, f.From)
+	}
+	if f.To != "" {
+		datePred += " AND local_date <= ?"
+		dateArgs = append(dateArgs, f.To)
+	}
 	query := fmt.Sprintf(`
 		WITH usage_raw AS (
 			%s
@@ -1976,41 +1974,33 @@ func duckUsageCTE(f db.UsageFilter, sessionID string) (string, []any) {
 					ELSE 'row:' || session_id || ':' || source || ':' ||
 						COALESCE(CAST(message_ordinal AS VARCHAR), '') || ':' ||
 						CAST(ts AS VARCHAR) || ':' || model
-				END AS dedup_group
+				END AS dedup_group,
+				%s AS local_date
 			FROM usage_raw
 		),
+		usage_windowed AS (
+			SELECT *
+			FROM usage_normalized
+			WHERE %s
+		),
 		usage_ranked AS (
 			SELECT *,
 				ROW_NUMBER() OVER (
 					PARTITION BY dedup_group
 					ORDER BY ts ASC, session_id ASC, COALESCE(message_ordinal, -1) ASC
 				) AS dedup_rank
-			FROM usage_normalized
+			FROM usage_windowed
 		),
 		usage_localized AS (
-			SELECT *,
-				%s AS local_date
+			SELECT *
 			FROM usage_ranked
 			WHERE dedup_rank = 1
-		)`, rawSQL, localDateSQL)
+		)`, rawSQL, localDateSQL, datePred)
 	args = append(args, localDateArg)
+	args = append(args, dateArgs...)
 	return query, args
 }
 
-func appendDuckUsageLocalDateFilter(
-	where string, args []any, f db.UsageFilter,
-) (string, []any) {
-	if f.From != "" {
-		where += "\n\t\tAND local_date >= ?"
-		args = append(args, f.From)
-	}
-	if f.To != "" {
-		where += "\n\t\tAND local_date <= ?"
-		args = append(args, f.To)
-	}
-	return where, args
-}
-
 type duckUsageBucket struct {
 	inputTok  int
 	outputTok int
@@ -2064,8 +2054,6 @@ func (s *Store) dailyUsageAggregateRows(
 	ctx context.Context, f db.UsageFilter,
 ) ([]duckUsageAggregateRow, error) {
 	cte, args := duckUsageCTE(f, "")
-	where := "TRUE"
-	where, args = appendDuckUsageLocalDateFilter(where, args, f)
 	query := cte + `
 		SELECT local_date, project, agent, model,
 			SUM(input_tokens_norm) AS input_tokens,
@@ -2078,7 +2066,6 @@ func (s *Store) dailyUsageAggregateRows(
 			SUM(CASE WHEN cost_usd IS NULL THEN cache_read_norm ELSE 0 END) AS billable_cache_read_tokens,
 			COALESCE(SUM(cost_usd), 0) AS explicit_cost
 		FROM usage_localized
-		WHERE ` + where + `
 		GROUP BY local_date, project, agent, model
 		ORDER BY local_date ASC, project ASC, agent ASC, model ASC`
 	rows, err := s.duck.QueryContext(ctx, query, args...)
@@ -2279,8 +2266,6 @@ func (s *Store) sessionUsageAggregateRows(
 	ctx context.Context, f db.UsageFilter, sessionID string,
 ) ([]duckUsageAggregateRow, error) {
 	cte, args := duckUsageCTE(f, sessionID)
-	where := "TRUE"
-	where, args = appendDuckUsageLocalDateFilter(where, args, f)
 	query := cte + `
 		SELECT session_id, project, agent, model,
 			ANY_VALUE(display_name) AS display_name,
@@ -2295,7 +2280,6 @@ func (s *Store) sessionUsageAggregateRows(
 			SUM(CASE WHEN cost_usd IS NULL THEN cache_read_norm ELSE 0 END) AS billable_cache_read_tokens,
 			COALESCE(SUM(cost_usd), 0) AS explicit_cost
 		FROM usage_localized
-		WHERE ` + where + `
 		GROUP BY session_id, project, agent, model
 		ORDER BY session_id ASC, model ASC`
 	rows, err := s.duck.QueryContext(ctx, query, args...)

internal/duckdb/store_contract_test.go

@@ -197,10 +197,9 @@ func duckContractAnalyticsTrendsAndUsage(
 	require.NoError(t, err)
 	require.Equal(t, 1, tools.TotalCalls)
 
-	trends, err := store.GetTrendsTerms(ctx, filter, []db.TrendTermInput{{
-		Term:     "alpha",
-		Variants: []string{"alpha"},
-	}}, "week")
+	trendTerms, err := db.ParseTrendTerms([]string{"alpha"})
+	require.NoError(t, err)
+	trends, err := store.GetTrendsTerms(ctx, filter, trendTerms, "week")
 	require.NoError(t, err)
 	require.Equal(t, 1, trends.Series[0].Total)
 

internal/duckdb/store_test.go

@@ -374,10 +374,9 @@ func TestStoreAnalyticsUsageAndTrends(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, 2, signals.UnscoredSessions)
 
-	trends, err := store.GetTrendsTerms(ctx, filter, []db.TrendTermInput{{
-		Term:     "alpha",
-		Variants: []string{"alpha"},
-	}}, "week")
+	trendTerms, err := db.ParseTrendTerms([]string{"alpha"})
+	require.NoError(t, err)
+	trends, err := store.GetTrendsTerms(ctx, filter, trendTerms, "week")
 	require.NoError(t, err)
 	assert.Equal(t, 1, trends.Series[0].Total)
 
@@ -1391,15 +1390,14 @@ func TestTrendsTermsApplySessionFiltersAndSystemPrefixExclusion(t *testing.T) {
 	require.NoError(t, err)
 	store := NewStoreFromDB(syncer.DB())
 
+	trendTerms, err := db.ParseTrendTerms([]string{"seam"})
+	require.NoError(t, err)
 	trends, err := store.GetTrendsTerms(ctx, db.AnalyticsFilter{
 		From:     "2026-01-22",
 		To:       "2026-01-22",
 		Timezone: "UTC",
 		Project:  "alpha",
-	}, []db.TrendTermInput{{
-		Term:     "seam",
-		Variants: []string{"seam"},
-	}}, "day")
+	}, trendTerms, "day")
 	require.NoError(t, err)
 	require.Len(t, trends.Series, 1)
 	assert.Equal(t, 1, trends.Series[0].Total)
@@ -1558,6 +1556,94 @@ func TestUsageDedupesClaudeMessageIDs(t *testing.T) {
 	assert.Equal(t, []string{"claude-test"}, sessionUsage.Models)
 }
 
+func TestUsageDedupPrefersInRangeDuplicate(t *testing.T) {
+	ctx := context.Background()
+	local := newLocalDB(t)
+	require.NoError(t, local.UpsertModelPricing([]db.ModelPricing{{
+		ModelPattern:  "claude-test",
+		InputPerMTok:  3,
+		OutputPerMTok: 15,
+	}}))
+
+	before := syncMessage("duck-usage-edge-a", 0, "assistant", "before midnight", "2026-01-12T23:30:00.000Z")
+	before.ClaudeMessageID = "edge-message"
+	before.ClaudeRequestID = "edge-request"
+	after := syncMessage("duck-usage-edge-b", 0, "assistant", "after midnight", "2026-01-13T00:30:00.000Z")
+	after.ClaudeMessageID = "edge-message"
+	after.ClaudeRequestID = "edge-request"
+
+	_, err := local.WriteSessionBatchAtomic([]db.SessionBatchWrite{
+		{
+			Session:         syncSession("duck-usage-edge-a", "alpha", "edge a", "2026-01-12T23:30:00.000Z", 1),
+			Messages:        []db.Message{before},
+			DataVersion:     1,
+			ReplaceMessages: true,
+		},
+		{
+			Session:         syncSession("duck-usage-edge-b", "alpha", "edge b", "2026-01-13T00:30:00.000Z", 1),
+			Messages:        []db.Message{after},
+			DataVersion:     1,
+			ReplaceMessages: true,
+		},
+	})
+	require.NoError(t, err)
+
+	syncer := newTestSync(t, filepath.Join(t.TempDir(), "usage-edge.duckdb"), local, SyncOptions{})
+	_, err = syncer.Push(ctx, true, nil)
+	require.NoError(t, err)
+	store := NewStoreFromDB(syncer.DB())
+
+	// The duplicate before midnight is outside the window but inside
+	// the padded UTC bounds and sorts first by timestamp. It must not
+	// win the dedup and suppress the in-range duplicate.
+	got, err := store.GetDailyUsage(ctx, db.UsageFilter{
+		From: "2026-01-13", To: "2026-01-13", Timezone: "UTC",
+	})
+	require.NoError(t, err)
+	assert.Equal(t, 1, got.Totals.InputTokens)
+	assert.Equal(t, 2, got.Totals.OutputTokens)
+}
+
+func TestTrendsTermsWordBoundaryAndOverlapParity(t *testing.T) {
+	ctx := context.Background()
+	local := newLocalDB(t)
+	start := "2026-01-22T09:00:00.000Z"
+	content := "seam seams seamless testing test attest"
+	_, err := local.WriteSessionBatchAtomic([]db.SessionBatchWrite{{
+		Session: syncSession("duck-trend-parity", "alpha", "trend parity", start, 1),
+		Messages: []db.Message{
+			syncMessage("duck-trend-parity", 0, "user", content, start),
+		},
+		DataVersion:     1,
+		ReplaceMessages: true,
+	}})
+	require.NoError(t, err)
+	syncer := newTestSync(t, filepath.Join(t.TempDir(), "trends-parity.duckdb"), local, SyncOptions{})
+	_, err = syncer.Push(ctx, true, nil)
+	require.NoError(t, err)
+	store := NewStoreFromDB(syncer.DB())
+
+	terms, err := db.ParseTrendTerms([]string{"seam", "test|testing"})
+	require.NoError(t, err)
+	filter := db.AnalyticsFilter{
+		From: "2026-01-22", To: "2026-01-22", Timezone: "UTC",
+	}
+
+	got, err := store.GetTrendsTerms(ctx, filter, terms, "day")
+	require.NoError(t, err)
+	require.Len(t, got.Series, 2)
+	// Word-bounded: "seamless" does not count for "seam", and
+	// "testing" is not double-counted via its "test" substring.
+	assert.Equal(t, 2, got.Series[0].Total)
+	assert.Equal(t, 2, got.Series[1].Total)
+
+	want, err := local.GetTrendsTerms(ctx, filter, terms, "day")
+	require.NoError(t, err)
+	require.Len(t, want.Series, 2)
+	assert.Equal(t, want.Series[0].Total, got.Series[0].Total)
+	assert.Equal(t, want.Series[1].Total, got.Series[1].Total)
+}
+
 func TestDailyUsageBreakdownsAndCacheSavings(t *testing.T) {
 	ctx := context.Background()
 	local := newLocalDB(t)