From acb6859b06e421c6852a3ce19b0ef76443d258b5 Mon Sep 17 00:00:00 2001 From: Adam <2363879+adamdotdevin@users.noreply.github.com> Date: Mon, 28 Sep 2026 06:08:52 -0500 Subject: [PATCH] fix(stats): attribute usage from catalog identity (#51833) --- .../core/src/domain/catalog-identity.test.ts | 50 ++++++++++++ .../stats/core/src/domain/catalog-identity.ts | 70 +++++++++++++++++ packages/stats/core/src/domain/geo.ts | 49 +++++++++++- packages/stats/core/src/domain/home.test.ts | 36 ++++++++- packages/stats/core/src/domain/home.ts | 46 +++++++++-- .../stats/core/src/domain/inference.test.ts | 41 ++++++++++ packages/stats/core/src/domain/inference.ts | 78 ++++++++++++++----- .../core/src/domain/model-normalization.ts | 1 + packages/stats/core/src/domain/model.ts | 51 +++++++++++- packages/stats/core/src/stat-sync.ts | 49 ++++++++++-- 10 files changed, 437 insertions(+), 34 deletions(-) create mode 100644 packages/stats/core/src/domain/catalog-identity.test.ts create mode 100644 packages/stats/core/src/domain/catalog-identity.ts diff --git a/packages/stats/core/src/domain/catalog-identity.test.ts b/packages/stats/core/src/domain/catalog-identity.test.ts new file mode 100644 index 0000000000..39508f14a3 --- /dev/null +++ b/packages/stats/core/src/domain/catalog-identity.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, test } from "bun:test" +import { catalogIdentity } from "./catalog-identity" + +describe("stats catalog identity", () => { + test("resolves OpenCode offerings to canonical labs", () => { + const identity = catalogIdentity({ + models: { "meituan/longcat-2.5-preview": {} }, + providers: { + opencode: { models: { "longcat-2.5-preview-free": { canonical_model_id: "meituan/longcat-2.5-preview" } } }, + "opencode-go": { + models: { "longcat-2.5-preview-free": { canonical_model_id: "meituan/longcat-2.5-preview" } }, + }, + }, + }) + + expect(identity.offerings.get("opencode/longcat-2.5-preview-free")).toBe("meituan") + expect(identity.offerings.get("opencode-go/longcat-2.5-preview-free")).toBe("meituan") + expect(identity.models.get("longcat-2.5-preview")).toBe("meituan") + }) + + test("does not guess a lab for a name shared by different canonical models", () => { + const identity = catalogIdentity({ + models: { "lab-a/model": {}, "lab-b/model": {} }, + providers: { + opencode: { models: { "model-free": { canonical_model_id: "lab-a/model" } } }, + "opencode-go": { models: { model: { canonical_model_id: "lab-b/model" } } }, + }, + }) + + expect(identity.offerings.get("opencode/model-free")).toBe("lab-a") + expect(identity.offerings.get("opencode-go/model")).toBe("lab-b") + expect(identity.models.has("model")).toBe(false) + }) + + test("accepts provider IDs that are already canonical catalog IDs", () => { + const identity = catalogIdentity({ + models: { "opencode/direct-model": {} }, + providers: { opencode: { models: { "direct-model": {} } } }, + }) + + expect(identity.offerings.get("opencode/direct-model")).toBe("opencode") + expect(identity.models.get("direct-model")).toBe("opencode") + }) + + test("rejects a catalog without published canonical identities", () => { + expect(() => catalogIdentity({ models: {}, providers: { opencode: { models: { model: {} } } } })).toThrow( + "Model catalog has no canonical OpenCode offerings", + ) + }) +}) diff --git a/packages/stats/core/src/domain/catalog-identity.ts b/packages/stats/core/src/domain/catalog-identity.ts new file mode 100644 index 0000000000..2fb6c55b1b --- /dev/null +++ b/packages/stats/core/src/domain/catalog-identity.ts @@ -0,0 +1,70 @@ +import { statModel } from "./model-normalization" + +const CATALOG_URL = "https://models.opencode.ai/catalog.json" +const STATS_PROVIDERS = ["opencode", "opencode-go"] as const +const cache: { value?: CatalogIdentity; expiresAt?: number } = {} + +export type CatalogIdentity = { + offerings: ReadonlyMap + models: ReadonlyMap +} + +export async function loadCatalogIdentity() { + if (cache.value && (cache.expiresAt ?? 0) > Date.now()) return { catalog: cache.value, stale: false } + return fetch(CATALOG_URL, { signal: AbortSignal.timeout(10_000) }) + .then(async (response) => { + if (!response.ok) throw new Error(`Model catalog returned ${response.status}`) + return catalogIdentity(await response.json()) + }) + .then((value) => { + cache.value = value + cache.expiresAt = Date.now() + 5 * 60_000 + return { catalog: value, stale: false } + }) + .catch(() => ({ catalog: cache.value, stale: true })) +} + +export function catalogIdentity(value: unknown): CatalogIdentity { + if (!record(value) || !record(value.models) || !record(value.providers)) throw new Error("Invalid model catalog") + const models = value.models + const providers = value.providers + + const offerings = new Map() + const candidates = new Map>() + STATS_PROVIDERS.forEach((providerID) => { + const provider = providers[providerID] + if (!record(provider) || !record(provider.models)) return + Object.entries(provider.models).forEach(([modelID, model]) => { + if (!record(model)) return + const canonicalID = + typeof model.canonical_model_id === "string" + ? model.canonical_model_id + : modelID in models + ? modelID + : `${providerID}/${modelID}` in models + ? `${providerID}/${modelID}` + : undefined + if (!canonicalID || !(canonicalID in models)) return + const [lab, ...parts] = canonicalID.split("/") + if (!lab || parts.length === 0) return + const canonicalModel = parts.join("/") + offerings.set(`${providerID}/${modelID.toLowerCase()}`, lab) + ;[modelID, canonicalModel].forEach((name) => { + const normalized = statModel(name, undefined) + candidates.set(normalized, (candidates.get(normalized) ?? new Set()).add(lab)) + }) + }) + }) + if (offerings.size === 0) throw new Error("Model catalog has no canonical OpenCode offerings") + + return { + offerings, + models: new Map( + [...candidates.entries()].flatMap(([model, labs]) => (labs.size === 1 ? [[model, [...labs][0]!]] : [])), + ), + } +} + +function record(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} diff --git a/packages/stats/core/src/domain/geo.ts b/packages/stats/core/src/domain/geo.ts index 975e0b927d..e1c3367d78 100644 --- a/packages/stats/core/src/domain/geo.ts +++ b/packages/stats/core/src/domain/geo.ts @@ -56,6 +56,7 @@ export declare namespace GeoStatRepo { }) => Effect.Effect readonly upsert: (rows: GeoStatRow[]) => Effect.Effect readonly deleteRetiredDimensions: (rows: GeoStatRow[]) => Effect.Effect + readonly deleteUnknownDimensions: (rows: GeoStatRow[]) => Effect.Effect } } @@ -218,7 +219,53 @@ export class GeoStatRepo extends Context.Service [statPeriodKey(row), row.model, row.country].join("\u0000"))) + const stale = yield* Effect.tryPromise({ + try: () => + db + .select({ + id: geoStat.id, + grain: geoStat.grain, + period_key: geoStat.period_key, + dataset: geoStat.dataset, + tier: geoStat.tier, + client: geoStat.client, + source: geoStat.source, + model: geoStat.model, + country: geoStat.country, + }) + .from(geoStat) + .where( + and( + eq(geoStat.provider, "unknown"), + inArray(geoStat.grain, scope.grains), + inArray(geoStat.period_key, scope.periodKeys), + inArray(geoStat.dataset, scope.datasets), + inArray(geoStat.client, scope.clients), + inArray(geoStat.source, scope.sources), + inArray(geoStat.model, [...new Set(rows.map((row) => row.model ?? "all"))]), + ), + ), + catch: (cause) => DatabaseError.make({ cause }), + }) + const ids = stale + .filter((row) => replacements.has([statPeriodKey(row), row.model, row.country].join("\u0000"))) + .map((row) => row.id) + yield* Effect.forEach( + chunks(ids, UPSERT_CHUNK_SIZE), + (chunk) => + Effect.tryPromise({ + try: () => db.delete(geoStat).where(and(eq(geoStat.provider, "unknown"), inArray(geoStat.id, chunk))), + catch: (cause) => DatabaseError.make({ cause }), + }), + { discard: true }, + ) + }) + + return GeoStatRepo.of({ listDaily, listByPeriod, upsert, deleteRetiredDimensions, deleteUnknownDimensions }) }), ) } diff --git a/packages/stats/core/src/domain/home.test.ts b/packages/stats/core/src/domain/home.test.ts index f2f98168d6..7a480a2902 100644 --- a/packages/stats/core/src/domain/home.test.ts +++ b/packages/stats/core/src/domain/home.test.ts @@ -5,7 +5,7 @@ import type { RetentionMetricRow } from "./home" process.env.SST_RESOURCE_App = JSON.stringify({ name: "opencode", stage: "test" }) process.env.SST_RESOURCE_StatsDatabase = JSON.stringify({ url: "mysql://localhost/stats" }) -const { buildRetentionEntries, buildStatsHomeData } = await import("./home") +const { buildRetentionEntries, buildStatsHomeData, normalizeStatRows } = await import("./home") test("daily rankings use the latest day while weekly rankings retain seven days and their previous-period change", () => { const rows = Array.from({ length: 14 }, (_, index) => @@ -46,6 +46,40 @@ test("daily rankings use the latest day while weekly rankings retain seven days expect(rankings["2M"]).toEqual(rankings["1W"]) }) +describe("model usage attribution", () => { + const row = { + periodKey: "2026-09-27", + updatedAt: new Date("2026-09-28T00:00:00.000Z"), + tier: "Go", + provider: "unknown", + model: "longcat-2.5-preview", + sessions: 1, + uniqueUsers: 1, + inputTokens: 50, + outputTokens: 50, + reasoningTokens: 0, + cacheReadTokens: 0, + totalTokens: 100, + inputCostMicrocents: 0, + outputCostMicrocents: 0, + totalCostMicrocents: 0, + } + + test("shows historical LongCat usage under Meituan", () => { + expect(normalizeStatRows([row])).toMatchObject([{ provider: "meituan", model: "longcat-2.5-preview" }]) + }) + + test("prefers recomputed rows over stale unknown dimensions", () => { + expect(normalizeStatRows([row, { ...row, provider: "meituan" }])).toMatchObject([ + { provider: "meituan", totalTokens: 100 }, + ]) + }) + + test("keeps unknown usage when a different lab has the same model name", () => { + expect(normalizeStatRows([row, { ...row, provider: "another-lab" }])).toHaveLength(2) + }) +}) + describe("retention aggregates", () => { test("pools the latest seven weekly cohorts and ranks models above the sample floor", () => { const rows = [ diff --git a/packages/stats/core/src/domain/home.ts b/packages/stats/core/src/domain/home.ts index f9cdf101f1..273d467a87 100644 --- a/packages/stats/core/src/domain/home.ts +++ b/packages/stats/core/src/domain/home.ts @@ -2,7 +2,7 @@ import { Client } from "@planetscale/database" import { Effect } from "effect" import { Resource } from "sst/resource" import type { ModelStatMetric } from "./model" -import { statProvider } from "./model-normalization" +import { RETIRED_STAT_PROVIDERS, statProvider } from "./model-normalization" import { isMissingRetentionTable } from "./retention" import { DATA_SITE_TIERS, normalizeTier } from "./stat" @@ -212,15 +212,20 @@ export function getStatsModelData( return Effect.tryPromise({ try: async () => { const [modelRows, retentionRows] = await Promise.all([listModelDaily(), listRetentionWeekly()]) - const normalized = modelRows.flatMap(normalizeStatRow) + const normalized = normalizeStatRows(modelRows) const resolvedModel = resolveModelName(model, normalized, provider) if (!resolvedModel) return null const window = modelRowsWindow(modelRows, "2M") const resolvedProvider = resolveModelProvider(resolvedModel, normalized, provider) + const countryRows = window + ? await listCountryTotals(window, { model: resolvedModel, provider: resolvedProvider }) + : [] return buildStatsModelData( resolvedModel, modelRows, - window ? await listCountryTotals(window, { model: resolvedModel, provider: resolvedProvider }) : [], + countryRows.length === 0 && window && resolvedProvider && resolvedProvider !== "unknown" + ? await listCountryTotals(window, { model: resolvedModel, provider: "unknown" }) + : countryRows, provider, retentionRows, ) @@ -383,7 +388,7 @@ export function buildStatsHomeData( countryRows: CountryTotalRow[], retentionRows: RetentionMetricRow[], ): StatsHomeData { - const normalized = modelRows.flatMap(normalizeStatRow) + const normalized = normalizeStatRows(modelRows) if (normalized.length === 0) return emptyStatsHomeData() const earliest = Math.min(...normalized.map((row) => row.periodStart)) @@ -444,7 +449,7 @@ function buildStatsModelData( providerParam?: string, retentionRows: RetentionMetricRow[] = [], ): StatsModelData | null { - const normalized = modelRows.flatMap(normalizeStatRow) + const normalized = normalizeStatRows(modelRows) if (normalized.length === 0) return null const model = resolveModelName(modelParam, normalized, providerParam) @@ -513,7 +518,7 @@ function buildStatsModelData( } function buildStatsLabData(providerParam: string, modelRows: ModelStatMetric[]): StatsLabData | null { - const normalized = modelRows.flatMap(normalizeStatRow) + const normalized = normalizeStatRows(modelRows) if (normalized.length === 0) return null const provider = resolveProviderName(providerParam, normalized) @@ -981,6 +986,30 @@ function createRangeRecord(value: (range: UsageRange) => T): Record row.provider !== "unknown" && !RETIRED_STAT_PROVIDERS.includes(row.provider.toLowerCase())) + .map((row) => [row.periodKey, normalizeTier(row.tier), row.model.toLowerCase(), row.provider].join("\u0000")), + ) + // Before cleanup finishes, an old unknown row may coexist with the lab row. + // Only suppress it when the known fallback identifies that same lab. + return rows + .filter( + (row) => + row.provider !== "unknown" || + !current.has( + [ + row.periodKey, + normalizeTier(row.tier), + row.model.toLowerCase(), + statProvider(row.model, "", "unknown"), + ].join("\u0000"), + ), + ) + .flatMap(normalizeStatRow) +} + function normalizeStatRow(row: ModelStatMetric): StatMetricRow[] { const periodStart = periodKeyTime(row.periodKey) const updatedAt = dateTime(row.updatedAt) @@ -991,7 +1020,10 @@ function normalizeStatRow(row: ModelStatMetric): StatMetricRow[] { periodStart, updatedAt, tier: normalizeTier(row.tier), - provider: statProvider(row.model, undefined, row.provider) || "unknown", + provider: + row.provider === "unknown" || RETIRED_STAT_PROVIDERS.includes(row.provider.toLowerCase()) + ? statProvider(row.model, undefined, row.provider) || "unknown" + : row.provider, model: row.model || "unknown", }, ] diff --git a/packages/stats/core/src/domain/inference.test.ts b/packages/stats/core/src/domain/inference.test.ts index 1e06f0e545..7e9ac065b2 100644 --- a/packages/stats/core/src/domain/inference.test.ts +++ b/packages/stats/core/src/domain/inference.test.ts @@ -14,6 +14,7 @@ import { statModel, statProvider, } from "./model-normalization" +import { catalogIdentity } from "./catalog-identity" describe("inference stat normalization", () => { test("normalizes model suffixes used by router/provider variants", () => { @@ -36,6 +37,7 @@ describe("inference stat normalization", () => { expect(modelAuthor("grok-build-0.1")).toBe("xai") expect(modelAuthor("hy3-preview")).toBe("tencent") expect(modelAuthor("kimi-k2.6")).toBe("moonshot") + expect(modelAuthor("longcat-2.5-preview")).toBe("meituan") expect(modelAuthor("mimo-v2-omni")).toBe("xiaomi") expect(modelAuthor("minimax-m2.7")).toBe("minimax") expect(modelAuthor("muse-spark-1.2-contributor")).toBe("meta") @@ -178,6 +180,45 @@ describe("inference stat normalization", () => { ]) }) + test("uses catalog labs in SQL and retains them in aggregate conversion", () => { + const catalog = catalogIdentity({ + models: { "meituan/longcat-2.5-preview": {}, "cohere/north-mini-code": {}, "zhipuai/glm-5.3": {} }, + providers: { + opencode: { + models: { + "longcat-2.5-preview-free": { canonical_model_id: "meituan/longcat-2.5-preview" }, + "north-mini-code": { canonical_model_id: "cohere/north-mini-code" }, + "glm-5.3": { canonical_model_id: "zhipuai/glm-5.3" }, + }, + }, + }, + }) + const [query] = buildStatsQueries( + new Date("2026-09-27"), + new Date("2026-09-28"), + { + namespace: "inference", + table: "generation", + dataset: "zen", + }, + catalog, + ) + + expect(query).toContain( + "WHEN lower(raw_provider) = 'opencode' AND lower(raw_model) = 'longcat-2.5-preview-free' THEN 'meituan'", + ) + expect(query).toContain("WHEN lower(model) = 'longcat-2.5-preview' THEN 'meituan'") + expect(query).toContain("WHEN lower(model) = 'north-mini-code' THEN 'cohere'") + expect(query).toContain("WHEN strpos(lower(model), 'glm') > 0 THEN 'zhipu'") + expect(query).not.toContain("WHEN lower(model) = 'glm-5.3' THEN 'zhipuai'") + expect(toModelAggregate(aggregate("longcat-2.5-preview", "meituan"), catalog)).toMatchObject([ + { model: "longcat-2.5-preview", provider: "meituan" }, + ]) + expect(toGeoAggregate({ ...aggregate("longcat-2.5-preview", "meituan"), country: "US" }, catalog)).toMatchObject([ + { model: "longcat-2.5-preview", provider: "meituan", country: "US" }, + ]) + }) + test("provider aggregates never keep opencode as the provider", () => { expect(toProviderAggregate({ ...aggregate("big-pickle", "opencode"), provider_model: "gpt-5" })).toMatchObject([ { provider: "openai" }, diff --git a/packages/stats/core/src/domain/inference.ts b/packages/stats/core/src/domain/inference.ts index fb45188aad..ee724ca6c8 100644 --- a/packages/stats/core/src/domain/inference.ts +++ b/packages/stats/core/src/domain/inference.ts @@ -1,5 +1,6 @@ import { Resource } from "sst/resource" import type { R2SqlData } from "../r2-sql" +import type { CatalogIdentity } from "./catalog-identity" import type { GeoStatAggregate } from "./geo" import type { ModelStatAggregate } from "./model" import { @@ -39,18 +40,28 @@ const LIVE_SOURCE_START = "2026-08-11T10:57:48.186Z" // R2 SQL results are cursor-paginated after aggregation. Separate usage and geo // queries per day/week avoid combining costly distinct user/session aggregates // with the high-cardinality geo dimensions. -export function buildStatsQueries(periodStart: Date, periodEnd: Date, input?: StatsQuerySource) { +export function buildStatsQueries( + periodStart: Date, + periodEnd: Date, + input?: StatsQuerySource, + catalog?: CatalogIdentity, +) { const source = input ?? { namespace: Resource.R2Sql.namespace, table: Resource.R2Sql.table, dataset: Resource.StatsSyncConfig.dataset, } return [...statPeriods("week", periodStart, periodEnd), ...statPeriods("day", periodStart, periodEnd)].flatMap( - (period) => [buildStatsQuery(period, source, "usage"), buildStatsQuery(period, source, "geo")], + (period) => [buildStatsQuery(period, source, "usage", catalog), buildStatsQuery(period, source, "geo", catalog)], ) } -export function buildRetentionQueries(periodStart: Date, periodEnd: Date, input?: StatsQuerySource): RetentionQuery[] { +export function buildRetentionQueries( + periodStart: Date, + periodEnd: Date, + input?: StatsQuerySource, + catalog?: CatalogIdentity, +): RetentionQuery[] { const source = input ?? { namespace: Resource.R2Sql.namespace, table: Resource.R2Sql.table, @@ -61,13 +72,14 @@ export function buildRetentionQueries(periodStart: Date, periodEnd: Date, input? // Combining the entire display window makes full syncs much more expensive. return periods.map((period) => ({ cohortDates: [period.start.toISOString().slice(0, 10)], - query: buildRetentionQuery([period], source), + query: buildRetentionQuery([period], source, catalog), })) } function buildRetentionQuery( periods: { start: Date; end: Date; returnStart: Date; returnEnd: Date }[], source: StatsQuerySource, + catalog?: CatalogIdentity, ) { const first = periods[0] const last = periods.at(-1)! @@ -104,6 +116,7 @@ ${periods WITH normalized AS ( SELECT ${activityWeekSql} AS activity_week, + model_requested AS raw_model, ${statModelSql("model_requested", "route_model")} AS model, COALESCE(NULLIF(route_model, ''), '') AS provider_model, COALESCE(NULLIF(provider_id, ''), '') AS raw_provider, @@ -125,7 +138,7 @@ WITH normalized AS ( ), filtered AS ( SELECT activity_week, - ${statProviderSql("model", "provider_model", "raw_provider")} AS provider, + ${statProviderSql("model", "provider_model", "raw_provider", "raw_model", catalog)} AS provider, model, user_key FROM normalized @@ -186,6 +199,7 @@ function buildStatsQuery( period: { grain: "day" | "week"; key: string; start: Date; end: Date }, source: StatsQuerySource, family: StatsQueryFamily, + catalog?: CatalogIdentity, ) { const periodStartValue = sqlString(period.start.toISOString()) const periodEndValue = sqlString(period.end.toISOString()) @@ -292,7 +306,7 @@ WITH normalized AS ( THEN 'Free' ELSE 'Go' END AS tier, - ${statProviderSql("model", "provider_model", "raw_provider")} AS provider, + ${statProviderSql("model", "provider_model", "raw_provider", "raw_model", catalog)} AS provider, provider_model, model, country, @@ -330,9 +344,9 @@ GROUP BY GROUPING SETS ( ` } -export function toModelAggregate(data: R2SqlData): ModelStatAggregate[] { +export function toModelAggregate(data: R2SqlData, catalog?: CatalogIdentity): ModelStatAggregate[] { const model = statModel(data.model, data.provider_model) - const provider = statProvider(model, data.provider_model, data.provider) + const provider = catalog ? data.provider || "unknown" : statProvider(model, data.provider_model, data.provider) if (!provider) return [] return toStatBaseAggregate(data).flatMap((base) => [ @@ -340,17 +354,20 @@ export function toModelAggregate(data: R2SqlData): ModelStatAggregate[] { ]) } -export function toProviderAggregate(data: R2SqlData): ProviderStatAggregate[] { - return toStatBaseAggregate(data).flatMap((base) => [ - { ...base, provider: statProvider(data.model, data.provider_model, data.provider) || "unknown" }, - ]) -} - -export function toGeoAggregate(data: R2SqlData): GeoStatAggregate[] { +export function toProviderAggregate(data: R2SqlData, catalog?: CatalogIdentity): ProviderStatAggregate[] { return toStatBaseAggregate(data).flatMap((base) => [ { ...base, - provider: statProvider(data.model, data.provider_model, data.provider) || "all", + provider: (catalog ? data.provider : statProvider(data.model, data.provider_model, data.provider)) || "unknown", + }, + ]) +} + +export function toGeoAggregate(data: R2SqlData, catalog?: CatalogIdentity): GeoStatAggregate[] { + return toStatBaseAggregate(data).flatMap((base) => [ + { + ...base, + provider: (catalog ? data.provider : statProvider(data.model, data.provider_model, data.provider)) || "all", model: statModel(data.model || "all", data.provider_model), country: normalizeCountry(data.country), continent: data.continent || "", @@ -358,14 +375,14 @@ export function toGeoAggregate(data: R2SqlData): GeoStatAggregate[] { ]) } -export function toRetentionAggregate(data: R2SqlData): RetentionStatAggregate[] { +export function toRetentionAggregate(data: R2SqlData, catalog?: CatalogIdentity): RetentionStatAggregate[] { if (!data.cohort_date || !data.model) return [] return [ { cohortDate: data.cohort_date, dataset: data.dataset || Resource.StatsSyncConfig.dataset, tier: data.tier || "all", - provider: statProvider(data.model, "", data.provider) || "unknown", + provider: (catalog ? data.provider : statProvider(data.model, "", data.provider)) || "unknown", model: statModel(data.model, undefined), eligibleUsers: integer(data, "eligible_users"), retainedUsers: integer(data, "retained_users"), @@ -485,9 +502,32 @@ function freeTierSql(tier: string, model: string) { OR lower(${model}) LIKE '%-free:global'` } -function statProviderSql(model: string, providerModel: string, provider: string) { +function statProviderSql( + model: string, + providerModel: string, + provider: string, + rawModel: string, + catalog?: CatalogIdentity, +) { + // Preserve existing lab dimensions when a rule disagrees with the catalog; + // changing them requires a separate historical aggregate migration. + const compatible = (name: string, lab: string) => { + const existing = statProvider(name, "", "unknown") + return existing === "unknown" || existing === lab + } return `CASE WHEN lower(${model}) IN (${[...STEALTH_MODELS].map(sqlString).join(", ")}) THEN 'unknown' +${[...(catalog?.offerings ?? [])] + .filter(([id, lab]) => compatible(statModel(id.slice(id.indexOf("/") + 1), undefined), lab)) + .map(([id, lab]) => { + const slash = id.indexOf("/") + return ` WHEN lower(${provider}) = ${sqlString(id.slice(0, slash))} AND lower(${rawModel}) = ${sqlString(id.slice(slash + 1))} THEN ${sqlString(lab)}` + }) + .join("\n")} +${[...(catalog?.models ?? [])] + .filter(([name, lab]) => compatible(name, lab)) + .map(([name, lab]) => ` WHEN lower(${model}) = ${sqlString(name)} THEN ${sqlString(lab)}`) + .join("\n")} ${MODEL_AUTHOR_RULES.map((item) => ` WHEN strpos(lower(${providerModel}), ${sqlString(item.match)}) > 0 THEN ${sqlString(item.author)}`).join("\n")} ${MODEL_AUTHOR_RULES.map((item) => ` WHEN strpos(lower(${model}), ${sqlString(item.match)}) > 0 THEN ${sqlString(item.author)}`).join("\n")} WHEN ${provider} <> '' AND lower(${provider}) NOT IN (${RETIRED_STAT_PROVIDERS.map(sqlString).join(", ")}) THEN ${provider} diff --git a/packages/stats/core/src/domain/model-normalization.ts b/packages/stats/core/src/domain/model-normalization.ts index 227eef1e90..7fff9759d6 100644 --- a/packages/stats/core/src/domain/model-normalization.ts +++ b/packages/stats/core/src/domain/model-normalization.ts @@ -8,6 +8,7 @@ export const MODEL_AUTHOR_RULES = [ { match: "hy3", author: "tencent" }, { match: "hy4", author: "tencent" }, { match: "kimi", author: "moonshot" }, + { match: "longcat", author: "meituan" }, { match: "mimo", author: "xiaomi" }, { match: "minimax", author: "minimax" }, { match: "muse-spark", author: "meta" }, diff --git a/packages/stats/core/src/domain/model.ts b/packages/stats/core/src/domain/model.ts index ec16d65158..abbc8ccdea 100644 --- a/packages/stats/core/src/domain/model.ts +++ b/packages/stats/core/src/domain/model.ts @@ -47,6 +47,7 @@ export declare namespace ModelStatRepo { readonly lastSyncedAt: () => Effect.Effect readonly upsert: (rows: ModelStatRow[]) => Effect.Effect readonly deleteRetiredDimensions: (rows: ModelStatRow[]) => Effect.Effect + readonly deleteUnknownDimensions: (rows: ModelStatRow[]) => Effect.Effect } } @@ -203,7 +204,55 @@ export class ModelStatRepo extends Context.Service [statPeriodKey(row), row.model].join("\u0000"))) + const stale = yield* Effect.tryPromise({ + try: () => + db + .select({ + id: modelStat.id, + grain: modelStat.grain, + period_key: modelStat.period_key, + dataset: modelStat.dataset, + tier: modelStat.tier, + client: modelStat.client, + source: modelStat.source, + model: modelStat.model, + }) + .from(modelStat) + .where( + and( + eq(modelStat.provider, "unknown"), + inArray(modelStat.grain, scope.grains), + inArray(modelStat.period_key, scope.periodKeys), + inArray(modelStat.dataset, scope.datasets), + inArray(modelStat.client, scope.clients), + inArray(modelStat.source, scope.sources), + inArray(modelStat.model, [...new Set(rows.map((row) => row.model))]), + ), + ), + catch: (cause) => DatabaseError.make({ cause }), + }) + const ids = stale + .filter((row) => replacements.has([statPeriodKey(row), row.model].join("\u0000"))) + .map((row) => row.id) + yield* Effect.forEach( + chunks(ids, UPSERT_CHUNK_SIZE), + (chunk) => + Effect.tryPromise({ + try: () => + db.delete(modelStat).where(and(eq(modelStat.provider, "unknown"), inArray(modelStat.id, chunk))), + catch: (cause) => DatabaseError.make({ cause }), + }), + { discard: true }, + ) + }) + + return ModelStatRepo.of({ listDaily, lastSyncedAt, upsert, deleteRetiredDimensions, deleteUnknownDimensions }) }), ) } diff --git a/packages/stats/core/src/stat-sync.ts b/packages/stats/core/src/stat-sync.ts index 7cf53ae936..a5cad3963e 100644 --- a/packages/stats/core/src/stat-sync.ts +++ b/packages/stats/core/src/stat-sync.ts @@ -1,6 +1,7 @@ import { DateTime, Effect } from "effect" import { Resource } from "sst/resource" import { DatabaseError } from "./database" +import { loadCatalogIdentity } from "./domain/catalog-identity" import { GeoStatRepo, rowsFromAggregates as geoRowsFromAggregates } from "./domain/geo" import { buildRetentionQueries, @@ -11,6 +12,7 @@ import { toRetentionAggregate, } from "./domain/inference" import { ModelStatRepo, rowsFromAggregates as modelRowsFromAggregates } from "./domain/model" +import { statProvider } from "./domain/model-normalization" import { ProviderStatRepo, rowsFromAggregates as providerRowsFromAggregates } from "./domain/provider" import { RetentionStatRepo, rowsFromAggregates as retentionRowsFromAggregates } from "./domain/retention" import { startOfIsoWeek, startOfUtcDay } from "./domain/stat" @@ -44,10 +46,18 @@ export const syncStats: (options?: { const providerStats = yield* ProviderStatRepo const geoStats = yield* GeoStatRepo const retentionStats = yield* RetentionStatRepo + const identity = yield* Effect.promise(loadCatalogIdentity) + const catalog = identity.catalog + if (identity.stale) + yield* Effect.logWarning( + catalog + ? "model catalog unavailable; using cached identity" + : "model catalog unavailable; using legacy stats attribution", + ) yield* logRuntimeCheck() - const queries = buildStatsQueries(periodStart, periodEnd) + const queries = buildStatsQueries(periodStart, periodEnd, undefined, catalog) yield* Effect.logInfo( `stats sync started ${JSON.stringify({ full: options?.full ?? false, periodStart, periodEnd, queries: queries.length })}`, ) @@ -70,12 +80,16 @@ export const syncStats: (options?: { concurrency: 4, }, ).pipe(Effect.map((batches) => batches.flat())) - const modelRows = modelRowsFromAggregates(rows.filter((row) => row.dimension === "model").flatMap(toModelAggregate)) + const modelRows = modelRowsFromAggregates( + rows.filter((row) => row.dimension === "model").flatMap((row) => toModelAggregate(row, catalog)), + ) const providerRows = providerRowsFromAggregates( - rows.filter((row) => row.dimension === "provider").flatMap(toProviderAggregate), + rows.filter((row) => row.dimension === "provider").flatMap((row) => toProviderAggregate(row, catalog)), ) const geoRows = geoRowsFromAggregates( - rows.filter((row) => row.dimension === "geo" || row.dimension === "geo_model").flatMap(toGeoAggregate), + rows + .filter((row) => row.dimension === "geo" || row.dimension === "geo_model") + .flatMap((row) => toGeoAggregate(row, catalog)), ) const retentionAvailable = yield* retentionStats.available() const retentionQueries = retentionAvailable @@ -86,6 +100,8 @@ export const syncStats: (options?: { Math.max(startOfUtcDay(periodEnd).getTime() - RETENTION_INCREMENTAL_LOOKBACK_MS, STATS_DATA_START_MS), ), startOfUtcDay(periodEnd), + undefined, + catalog, ) : [] yield* Effect.logInfo(`stats sync querying retention ${JSON.stringify({ queries: retentionQueries.length })}`) @@ -106,7 +122,9 @@ export const syncStats: (options?: { ), ), { concurrency: 4 }, - ).pipe(Effect.map((batches) => batches.flatMap((batch) => batch.flatMap(toRetentionAggregate)))), + ).pipe( + Effect.map((batches) => batches.flatMap((batch) => batch.flatMap((row) => toRetentionAggregate(row, catalog)))), + ), ) yield* Effect.logInfo( @@ -136,6 +154,27 @@ export const syncStats: (options?: { ], { concurrency: "unbounded", discard: true }, ) + yield* Effect.all( + [ + modelStats.deleteUnknownDimensions( + modelRows.filter( + (row) => + row.provider !== "unknown" && + (catalog?.models.get(row.model) ?? statProvider(row.model, "", "unknown")) === row.provider, + ), + ), + geoStats.deleteUnknownDimensions( + geoRows.filter( + (row) => + row.model !== "all" && + row.provider !== "unknown" && + (catalog?.models.get(row.model ?? "all") ?? statProvider(row.model ?? "all", "", "unknown")) === + row.provider, + ), + ), + ], + { concurrency: "unbounded", discard: true }, + ) yield* Effect.logInfo( `stats sync complete ${JSON.stringify({