chore: merge dev into v2 (#36312)

Co-authored-by: opencode-agent[bot] <219766164+opencode-agent[bot]@users.noreply.github.com> Co-authored-by: LukeParkerDev <10430890+Hona@users.noreply.github.com> Co-authored-by: opencode-agent[bot] <opencode-agent[bot]@users.noreply.github.com> Co-authored-by: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Co-authored-by: Brendan Allan <14191578+Brendonovich@users.noreply.github.com> Co-authored-by: Aarav Sareen <96787824+arvsrn@users.noreply.github.com> Co-authored-by: Julian Coy <julian@ex-machina.co> Co-authored-by: Brendan Allan <git@brendonovich.dev> Co-authored-by: usrnk1 <7547651+usrnk1@users.noreply.github.com> Co-authored-by: opencode <opencode@sst.dev> Co-authored-by: Vladimir Glafirov <vglafirov@gitlab.com> Co-authored-by: Adam <2363879+adamdotdevin@users.noreply.github.com> Co-authored-by: Frank <frank@anoma.ly> Co-authored-by: Jay <53023+jayair@users.noreply.github.com> Co-authored-by: Dustin Deus <deusdustin@gmail.com> Co-authored-by: Kit Langton <kit.langton@gmail.com> Co-authored-by: James Long <longster@gmail.com> Co-authored-by: Simon Klee <hello@simonklee.dk> Co-authored-by: Jay <air@live.ca> Co-authored-by: Jack <jack@anoma.ly> Co-authored-by: David Hill <1879069+iamdavidhill@users.noreply.github.com> Co-authored-by: Aiden Cline <aidenpcline@gmail.com> Co-authored-by: James Long <jlongster@users.noreply.github.com> Co-authored-by: 冯基魁 <56265583+fengjikui@users.noreply.github.com> Co-authored-by: Aiden Cline <rekram1-node@users.noreply.github.com> Co-authored-by: Victor Navarro <vn4varro@gmail.com>
2026-07-10 15:11:29 -05:00
parent 9028c2d8f8
commit 43ecf3ff1b
82 changed files with 4520 additions and 1232 deletions
@@ -113,32 +113,32 @@ const poll: (
  return yield* poll(client, queryExecutionId, attempt + 1)
 })

-const results: (
-  client: AwsAthenaClient,
-  queryExecutionId: string,
-  nextToken?: string,
-) => Effect.Effect<AthenaData[], AthenaQueryError> = Effect.fn("Athena.results")(function* (
-  client: AwsAthenaClient,
-  queryExecutionId: string,
-  nextToken?: string,
-) {
-  const result = yield* Effect.tryPromise({
-    try: () =>
-      client.send(
-        new GetQueryResultsCommand({
-          QueryExecutionId: queryExecutionId,
-          NextToken: nextToken,
-          MaxResults: ATHENA_PAGE_SIZE,
-        }),
-      ),
-    catch: (cause) => new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }),
+const results: (client: AwsAthenaClient, queryExecutionId: string) => Effect.Effect<AthenaData[], AthenaQueryError> =
+  Effect.fn("Athena.results")(function* (client: AwsAthenaClient, queryExecutionId: string) {
+    // Accumulate pages iteratively; recursive spreads copied every previously
+    // fetched row per page and blew up memory on large result sets.
+    const rows: AthenaData[] = []
+    let nextToken: string | undefined
+    while (true) {
+      const result = yield* Effect.tryPromise({
+        try: () =>
+          client.send(
+            new GetQueryResultsCommand({
+              QueryExecutionId: queryExecutionId,
+              NextToken: nextToken,
+              MaxResults: ATHENA_PAGE_SIZE,
+            }),
+          ),
+        catch: (cause) =>
+          new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }),
+      })
+      const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? []
+      // The first page starts with the header row.
+      for (const row of (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1)) rows.push(rowData(columns, row))
+      if (!result.NextToken) return rows
+      nextToken = result.NextToken
+    }
  })
-  const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? []
-  const rows = (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1).map((row) => rowData(columns, row))
-
-  if (!result.NextToken) return rows
-  return [...rows, ...(yield* results(client, queryExecutionId, result.NextToken))]
-})

 function rowData(columns: string[], row: Row): AthenaData {
  return Object.fromEntries(
@@ -94,10 +94,15 @@ export type StatsModelComparisonEntry = {
  tokenShare: number
  tokenChange: number
  totals: StatsModelData["totals"]
+  usage: ModelUsagePoint[]
+}
+export type StatsModelComparisonInput = {
+  provider: string
+  model: string
 }
 export type StatsModelComparisonData = {
  updatedAt: string | null
-  models: [StatsModelComparisonEntry | null, StatsModelComparisonEntry | null]
+  models: (StatsModelComparisonEntry | null)[]
 }
 export type StatsHomeData = {
  updatedAt: string | null
@@ -289,27 +294,35 @@ function dateValue(value: unknown) {
  return value instanceof Date ? value : new Date(stringValue(value))
 }

-export const getStatsModelComparisonData: (
-  firstProvider: string,
-  firstModel: string,
-  secondProvider: string,
-  secondModel: string,
-) => Effect.Effect<StatsModelComparisonData, DatabaseError, ModelStatRepo> = Effect.fn("StatsModelComparison.getData")(
-  function* (firstProvider, firstModel, secondProvider, secondModel) {
+export const getStatsModelsComparisonData: (
+  models: readonly StatsModelComparisonInput[],
+) => Effect.Effect<StatsModelComparisonData, DatabaseError, ModelStatRepo> = Effect.fn("StatsModelsComparison.getData")(
+  function* (models) {
    const modelStats = yield* ModelStatRepo
    const rows = yield* modelStats.listDaily()
-    const first = toComparisonEntry(buildStatsModelData(firstModel, rows, [], firstProvider))
-    const second = toComparisonEntry(buildStatsModelData(secondModel, rows, [], secondProvider))
-    const latest = [first?.updatedAt, second?.updatedAt]
+    const entries = models.map((model) => toComparisonEntry(buildStatsModelData(model.model, rows, [], model.provider)))
+    const latest = entries
+      .map((model) => model?.updatedAt)
      .flatMap((value) => (value ? [dateTime(value)] : []))
      .toSorted((a, b) => b - a)[0]
    return {
      updatedAt: latest === undefined ? null : new Date(latest).toISOString(),
-      models: [first, second],
+      models: entries,
    }
  },
 )

+export const getStatsModelComparisonData = (
+  firstProvider: string,
+  firstModel: string,
+  secondProvider: string,
+  secondModel: string,
+) =>
+  getStatsModelsComparisonData([
+    { provider: firstProvider, model: firstModel },
+    { provider: secondProvider, model: secondModel },
+  ])
+
 function buildStatsHomeData(
  modelRows: ModelStatMetric[],
  providerRows: ProviderStatMetric[],
@@ -501,6 +514,7 @@ function toComparisonEntry(data: StatsModelData | null): StatsModelComparisonEnt
    tokenShare: data.tokenShare,
    tokenChange: data.tokenChange,
    totals: data.totals,
+    usage: data.usage,
  }
 }

@@ -14,7 +14,10 @@ import { normalizeCountry, normalizeTier, type StatBaseAggregate } from "./stat"

 export type StatDimension = "model" | "provider" | "geo" | "geo_model"

-export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
+// All stat dimensions and both grains are computed in one query via GROUPING SETS so
+// the source table is scanned once per sync pass; separate queries per dimension (and
+// the previous weekly/daily UNION ALL) each re-scanned the same events.
+export function buildStatsQuery(periodStart: Date, periodEnd: Date) {
  const periodStartValue = sqlString(periodStart.toISOString())
  const periodEndValue = sqlString(periodEnd.toISOString())
  const periodStartDateValue = sqlString(periodStart.toISOString().slice(0, 10))
@@ -22,23 +25,6 @@ export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: S
  const sourceTable = [Resource.InferenceEvent.catalog, Resource.InferenceEvent.database, Resource.InferenceEvent.table]
    .map(sqlIdentifier)
    .join(".")
-  const dimensionSql = (() => {
-    if (dimension === "model")
-      return {
-        select: "provider, model, COALESCE(MAX(NULLIF(provider_model, '')), '') AS provider_model",
-        groupBy: "provider, model",
-      }
-    if (dimension === "provider") return { select: "provider", groupBy: "provider" }
-    if (dimension === "geo_model")
-      return {
-        select: "provider, model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
-        groupBy: "provider, model, country",
-      }
-    return {
-      select: "'all' AS provider, 'all' AS model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
-      groupBy: "country",
-    }
-  })()
  const aggregateColumns = `
    COUNT(DISTINCT session) AS sessions,
    COUNT(*) AS requests,
@@ -135,34 +121,41 @@ WITH normalized AS (
    COALESCE(cost_total_microcents, cost_total * 1000000) AS cost_total_microcents
  FROM normalized
  WHERE lower(model) NOT IN (${[...EXCLUDED_MODELS].map(sqlString).join(", ")})
-), weekly AS (
+), periods AS (
  SELECT
    concat(CAST(year_of_week(event_time) AS varchar), '-W', lpad(CAST(week(event_time) AS varchar), 2, '0')) AS week_key,
+    substr(to_iso8601(date_trunc('day', event_time)), 1, 10) AS day_key,
    *
  FROM filtered
-), daily AS (
-  SELECT substr(to_iso8601(date_trunc('day', event_time)), 1, 10) AS day_key, *
-  FROM filtered
 )
 SELECT
-  'week' AS grain,
-  week_key AS period_key,
+  CASE WHEN grouping(week_key) = 0 THEN 'week' ELSE 'day' END AS grain,
+  COALESCE(week_key, day_key) AS period_key,
  ${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
+  CASE
+    WHEN grouping(country) = 0 AND grouping(model) = 0 THEN 'geo_model'
+    WHEN grouping(country) = 0 THEN 'geo'
+    WHEN grouping(model) = 0 THEN 'model'
+    ELSE 'provider'
+  END AS dimension,
  tier,
-  ${dimensionSql.select},
+  CASE WHEN grouping(provider) = 0 THEN provider ELSE 'all' END AS provider,
+  CASE WHEN grouping(model) = 0 THEN model WHEN grouping(country) = 0 THEN 'all' END AS model,
+  CASE WHEN grouping(model) = 0 AND grouping(country) = 1 THEN COALESCE(MAX(NULLIF(provider_model, '')), '') END AS provider_model,
+  CASE WHEN grouping(country) = 0 THEN country END AS country,
+  CASE WHEN grouping(country) = 0 THEN COALESCE(MAX(NULLIF(continent, '')), '') END AS continent,
  ${aggregateColumns}
-FROM weekly
-GROUP BY week_key, tier, ${dimensionSql.groupBy}
-UNION ALL
-SELECT
-  'day' AS grain,
-  day_key AS period_key,
-  ${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
-  tier,
-  ${dimensionSql.select},
-  ${aggregateColumns}
-FROM daily
-GROUP BY day_key, tier, ${dimensionSql.groupBy}
+FROM periods
+GROUP BY GROUPING SETS (
+  (week_key, tier, provider, model),
+  (week_key, tier, provider),
+  (week_key, tier, country),
+  (week_key, tier, provider, model, country),
+  (day_key, tier, provider, model),
+  (day_key, tier, provider),
+  (day_key, tier, country),
+  (day_key, tier, provider, model, country)
+)
 ORDER BY grain, period_key, total_tokens DESC
 `
 }
@@ -1,4 +1,4 @@
-import { and, asc, eq, inArray, or } from "drizzle-orm"
+import { and, asc, eq, inArray, max, or } from "drizzle-orm"
 import { Effect, Layer } from "effect"
 import * as Context from "effect/Context"
 import { DatabaseError, DrizzleClient } from "../database"
@@ -43,6 +43,7 @@ export type ModelStatMetric = {
 export declare namespace ModelStatRepo {
  export interface Service {
    readonly listDaily: () => Effect.Effect<ModelStatMetric[], DatabaseError>
+    readonly lastSyncedAt: () => Effect.Effect<Date | null, DatabaseError>
    readonly upsert: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
    readonly deleteRetiredDimensions: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
  }
@@ -111,6 +112,14 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
        })
      })

+      const lastSyncedAt = Effect.fn("ModelStatRepo.lastSyncedAt")(function* () {
+        const result = yield* Effect.tryPromise({
+          try: () => db.select({ value: max(modelStat.updated_at) }).from(modelStat),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+        return result[0]?.value ?? null
+      })
+
      const upsert = Effect.fn("ModelStatRepo.upsert")(function* (rows: ModelStatRow[]) {
        yield* Effect.forEach(
          chunks(rows, UPSERT_CHUNK_SIZE),
@@ -192,7 +201,7 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
        })
      })

-      return ModelStatRepo.of({ listDaily, upsert, deleteRetiredDimensions })
+      return ModelStatRepo.of({ listDaily, lastSyncedAt, upsert, deleteRetiredDimensions })
    }),
  )
 }
@@ -12,85 +12,90 @@ const DATALAKE_INGESTION_LAG_MS = 5 * 60_000
 const STATS_DATA_START_MS = new Date("2026-05-28T00:00:00.000Z").getTime()
 const WEEK_MS = 7 * 86_400_000
 const DISPLAY_WINDOW_MS = 56 * 86_400_000
+// Anchor incremental passes to the ISO week containing this lookback, so the pass
+// after a week boundary still recomputes the previous week's final aggregates even
+// if the boundary pass itself failed.
+const INCREMENTAL_LOOKBACK_MS = 2 * 3_600_000

 export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string }
 export type SyncStatsError = AthenaQueryError | AthenaQueryTimeoutError | DatabaseError

-export const syncStats: () => Effect.Effect<
-  SyncStatsResult,
-  SyncStatsError,
-  Athena | ModelStatRepo | ProviderStatRepo | GeoStatRepo
-> = Effect.fn("StatSync.sync")(function* () {
-  const startedAt = yield* DateTime.nowAsDate
-  const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000)
-  // May 27 was partial, so keep Athena stats anchored at the first complete day.
-  const periodStart = new Date(
+export const syncStats: (options?: {
+  full?: boolean
+}) => Effect.Effect<SyncStatsResult, SyncStatsError, Athena | ModelStatRepo | ProviderStatRepo | GeoStatRepo> =
+  Effect.fn("StatSync.sync")(function* (options?: { full?: boolean }) {
+    const startedAt = yield* DateTime.nowAsDate
+    const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000)
+    const periodStart = options?.full ? fullPeriodStart(periodEnd) : incrementalPeriodStart(periodEnd)
+    const athena = yield* Athena
+    const modelStats = yield* ModelStatRepo
+    const providerStats = yield* ProviderStatRepo
+    const geoStats = yield* GeoStatRepo
+
+    yield* logRuntimeCheck()
+
+    const rows = yield* athena.query(buildStatsQuery(periodStart, periodEnd))
+    const modelRows = modelRowsFromAggregates(rows.filter((row) => row.dimension === "model").flatMap(toModelAggregate))
+    const providerRows = providerRowsFromAggregates(
+      rows.filter((row) => row.dimension === "provider").flatMap(toProviderAggregate),
+    )
+    const geoRows = geoRowsFromAggregates(
+      rows.filter((row) => row.dimension === "geo" || row.dimension === "geo_model").flatMap(toGeoAggregate),
+    )
+
+    yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
+      concurrency: "unbounded",
+      discard: true,
+    })
+    yield* Effect.all(
+      [
+        modelStats.deleteRetiredDimensions(modelRows),
+        providerStats.deleteRetiredDimensions(providerRows),
+        geoStats.deleteRetiredDimensions(geoRows),
+      ],
+      { concurrency: "unbounded", discard: true },
+    )
+
+    yield* Effect.logInfo(
+      `stats sync complete ${JSON.stringify({
+        startedAt: startedAt.toISOString(),
+        periodStart: periodStart.toISOString(),
+        periodEnd: periodEnd.toISOString(),
+        rows: modelRows.length,
+        providerRows: providerRows.length,
+        geoRows: geoRows.length,
+        stage: Resource.App.stage,
+      })}`,
+    )
+
+    return {
+      ok: true,
+      rows: modelRows.length,
+      startedAt: startedAt.toISOString(),
+      periodStart: periodStart.toISOString(),
+      periodEnd: periodEnd.toISOString(),
+    }
+  })
+
+// May 27 was partial, so keep Athena stats anchored at the first complete day.
+function fullPeriodStart(periodEnd: Date) {
+  return new Date(
    Math.max(
      Math.min(startOfIsoWeek(periodEnd).getTime() - WEEK_MS, periodEnd.getTime() - DISPLAY_WINDOW_MS),
      STATS_DATA_START_MS,
    ),
  )
-  const athena = yield* Athena
-  const modelStats = yield* ModelStatRepo
-  const providerStats = yield* ProviderStatRepo
-  const geoStats = yield* GeoStatRepo
+}

-  yield* logRuntimeCheck()
-
-  const [modelAggregates, providerAggregates, geoAggregates, geoModelAggregates] = yield* Effect.all(
-    [
-      athena
-        .query(buildStatsQuery(periodStart, periodEnd, "model"))
-        .pipe(Effect.map((rows) => rows.flatMap(toModelAggregate))),
-      athena
-        .query(buildStatsQuery(periodStart, periodEnd, "provider"))
-        .pipe(Effect.map((rows) => rows.flatMap(toProviderAggregate))),
-      athena
-        .query(buildStatsQuery(periodStart, periodEnd, "geo"))
-        .pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
-      athena
-        .query(buildStatsQuery(periodStart, periodEnd, "geo_model"))
-        .pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
-    ],
-    { concurrency: "unbounded" },
+// Events are append-only, so completed periods never change once synced; hourly
+// passes only recompute the periods the current ISO week can still touch. The daily
+// full pass refreshes the whole display window (normalization changes, retired
+// dimension cleanup).
+function incrementalPeriodStart(periodEnd: Date) {
+  return new Date(
+    Math.max(startOfIsoWeek(new Date(periodEnd.getTime() - INCREMENTAL_LOOKBACK_MS)).getTime(), STATS_DATA_START_MS),
  )
-  const modelRows = modelRowsFromAggregates(modelAggregates)
-  const providerRows = providerRowsFromAggregates(providerAggregates)
-  const geoRows = geoRowsFromAggregates([...geoAggregates, ...geoModelAggregates])
-
-  yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
-    concurrency: "unbounded",
-    discard: true,
-  })
-  yield* Effect.all(
-    [
-      modelStats.deleteRetiredDimensions(modelRows),
-      providerStats.deleteRetiredDimensions(providerRows),
-      geoStats.deleteRetiredDimensions(geoRows),
-    ],
-    { concurrency: "unbounded", discard: true },
-  )
-
-  yield* Effect.logInfo(
-    `stats sync complete ${JSON.stringify({
-      startedAt: startedAt.toISOString(),
-      periodStart: periodStart.toISOString(),
-      periodEnd: periodEnd.toISOString(),
-      rows: modelRows.length,
-      providerRows: providerRows.length,
-      geoRows: geoRows.length,
-      stage: Resource.App.stage,
-    })}`,
-  )
-
-  return {
-    ok: true,
-    rows: modelRows.length,
-    startedAt: startedAt.toISOString(),
-    periodStart: periodStart.toISOString(),
-    periodEnd: periodEnd.toISOString(),
-  }
-})
+}

 function logRuntimeCheck() {
  return Effect.logInfo(