CloudAxe/cloudaxe-opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 6

fix(stats): attribute usage from catalog identity (#51833)

Browse source
This commit is contained in:
Adam 2026-09-28 06:08:52 -05:00 committed by GitHub
parent 9f9e73ae0f
commit acb6859b06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 437 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
packages/stats/core/src/domain/catalog-identity.test.ts Normal file
View file

@ -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",
)
})
})

70
packages/stats/core/src/domain/catalog-identity.ts Normal file
View file

@ -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<string, string>
models: ReadonlyMap<string, string>
}
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<string, string>()
const candidates = new Map<string, Set<string>>()
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<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
}

49
packages/stats/core/src/domain/geo.ts
View file

@ -56,6 +56,7 @@ export declare namespace GeoStatRepo {
}) => Effect.Effect<GeoStatRow[], DatabaseError>
readonly upsert: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
readonly deleteRetiredDimensions: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
readonly deleteUnknownDimensions: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
}
}
@ -218,7 +219,53 @@ export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Servic
})
})
return GeoStatRepo.of({ listDaily, listByPeriod, upsert, deleteRetiredDimensions })
const deleteUnknownDimensions = Effect.fn("GeoStatRepo.deleteUnknownDimensions")(function* (rows: GeoStatRow[]) {
const scope = statRowScope(rows)
if (!scope) return
const replacements = new Set(rows.map((row) => [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 })
}),
)
}

36
packages/stats/core/src/domain/home.test.ts
View file

@ -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 = [

46
packages/stats/core/src/domain/home.ts
View file

@ -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<T>(value: (range: UsageRange) => T): Record<UsageRang
}
}
export function normalizeStatRows(rows: ModelStatMetric[]) {
const current = new Set(
rows
.filter((row) => 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",
},
]

41
packages/stats/core/src/domain/inference.test.ts
View file

@ -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" },

78
packages/stats/core/src/domain/inference.ts
View file

@ -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}

1
packages/stats/core/src/domain/model-normalization.ts
View file

@ -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" },

51
packages/stats/core/src/domain/model.ts
View file

@ -47,6 +47,7 @@ export declare namespace ModelStatRepo {
readonly lastSyncedAt: () => Effect.Effect<Date | null, DatabaseError>
readonly upsert: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
readonly deleteRetiredDimensions: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
readonly deleteUnknownDimensions: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
}
}
@ -203,7 +204,55 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
})
})
return ModelStatRepo.of({ listDaily, lastSyncedAt, upsert, deleteRetiredDimensions })
const deleteUnknownDimensions = Effect.fn("ModelStatRepo.deleteUnknownDimensions")(function* (
rows: ModelStatRow[],
) {
const scope = statRowScope(rows)
if (!scope) return
const replacements = new Set(rows.map((row) => [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 })
}),
)
}

49
packages/stats/core/src/stat-sync.ts
View file

@ -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({