From 3e491df445ca927297368dad286f384a8cdba989 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 17 Apr 2026 23:47:07 +0000 Subject: [PATCH] =?UTF-8?q?feat(llm-registry):=20phase=200/1=20=E2=80=94?= =?UTF-8?q?=20schema,=20pub/sub,=20OTEL=20counter,=20CI=20drift=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 0 of the LLM model registry productionization pipeline: - non-blocking sync-prices:check job wired into pr_checks.yml - OTEL counter llm_missing_model_enrichment{gen_ai_system, has_provider_cost} emitted from enrichCreatableEvents when the registry is loaded but a span can't be priced Phase 1 schema + registry wiring (additive, no breaking changes): - New LlmModel columns: resolvedAt, needsReview, releaseDate, deprecationDate, knowledgeCutoff, supportsStructuredOutput, supportsParallelToolCalls, supportsStreamingToolCalls - ModelPricingRegistry.loadFromDatabase filters needsReview=true rows so auto-priced and freshly-synced rows stay out of the live registry until admin approval - New LlmRegistryPubSub service on channel llm-registry:reload so webapp replicas pick up DB changes within seconds (5-min periodic reload kept as backstop) - llmPricingRegistry subscribes on boot; publishLlmRegistryReload() exposed for admin routes + future cloud-repo trigger.dev tasks - Admin API (create/update/delete/reload/seed) and admin UI routes now publish on the channel after mutations and accept the widened source enum (langfuse, auto, research, provider-api) + the new review/metadata fields Co-Authored-By: Eric Allam --- .github/workflows/pr_checks.yml | 4 + .github/workflows/sync-prices-check.yml | 45 ++++++++ .../admin.api.v1.llm-models.$modelId.ts | 62 +++++++++- .../routes/admin.api.v1.llm-models.reload.ts | 12 +- .../routes/admin.api.v1.llm-models.seed.ts | 7 +- .../app/routes/admin.api.v1.llm-models.ts | 62 +++++++++- .../app/routes/admin.llm-models.$modelId.tsx | 7 +- .../app/routes/admin.llm-models._index.tsx | 9 +- .../app/routes/admin.llm-models.new.tsx | 6 +- .../app/v3/llmPricingRegistry.server.ts | 44 ++++++- .../v3/services/llmRegistryPubSub.server.ts | 109 ++++++++++++++++++ .../v3/utils/enrichCreatableEvents.server.ts | 47 ++++++++ .../migration.sql | 9 ++ .../database/prisma/schema.prisma | 38 ++++-- .../llm-model-catalog/src/registry.ts | 9 +- 15 files changed, 447 insertions(+), 23 deletions(-) create mode 100644 .github/workflows/sync-prices-check.yml create mode 100644 apps/webapp/app/v3/services/llmRegistryPubSub.server.ts create mode 100644 internal-packages/database/prisma/migrations/20260417233744_add_llm_model_catalog_metadata/migration.sql diff --git a/.github/workflows/pr_checks.yml b/.github/workflows/pr_checks.yml index dab18223e..171b877e2 100644 --- a/.github/workflows/pr_checks.yml +++ b/.github/workflows/pr_checks.yml @@ -33,3 +33,7 @@ jobs: sdk-compat: uses: ./.github/workflows/sdk-compat.yml secrets: inherit + + sync-prices-check: + uses: ./.github/workflows/sync-prices-check.yml + secrets: inherit diff --git a/.github/workflows/sync-prices-check.yml b/.github/workflows/sync-prices-check.yml new file mode 100644 index 000000000..33bc1073b --- /dev/null +++ b/.github/workflows/sync-prices-check.yml @@ -0,0 +1,45 @@ +name: "💰 LLM Prices Drift" + +# Non-blocking drift check for the Langfuse-sourced LLM prices JSON. Runs on +# every PR; failure here is a warning only (does not block merge) and is there +# to prompt a human to run `pnpm run sync-prices` when the upstream JSON has +# moved ahead of our checked-in copy. Phase 2 of the productionization pipeline +# replaces this with an hourly trigger.dev task that writes directly to +# Postgres, at which point this workflow can be removed. + +on: + workflow_call: + +permissions: + contents: read + +jobs: + sync-prices-check: + name: sync-prices drift (non-blocking) + runs-on: ubuntu-latest + # IMPORTANT: non-blocking — do not remove. This job is advisory only while + # the productionization pipeline is being rolled out. + continue-on-error: true + + steps: + - name: ⬇️ Checkout repo + uses: actions/checkout@v4 + with: + fetch-depth: 1 + + - name: ⎔ Setup pnpm + uses: pnpm/action-setup@v4 + with: + version: 10.23.0 + + - name: ⎔ Setup node + uses: buildjet/setup-node@v4 + with: + node-version: 20.20.0 + cache: "pnpm" + + - name: 📥 Download deps + run: pnpm install --frozen-lockfile + + - name: 💰 Check Langfuse prices drift + run: pnpm run --filter @internal/llm-model-catalog sync-prices:check diff --git a/apps/webapp/app/routes/admin.api.v1.llm-models.$modelId.ts b/apps/webapp/app/routes/admin.api.v1.llm-models.$modelId.ts index e473c2c72..ca56edbe2 100644 --- a/apps/webapp/app/routes/admin.api.v1.llm-models.$modelId.ts +++ b/apps/webapp/app/routes/admin.api.v1.llm-models.$modelId.ts @@ -2,6 +2,7 @@ import { type ActionFunctionArgs, type LoaderFunctionArgs, json } from "@remix-r import { z } from "zod"; import { prisma } from "~/db.server"; import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; +import { publishLlmRegistryReload } from "~/v3/llmPricingRegistry.server"; export async function loader({ request, params }: LoaderFunctionArgs) { await requireAdminApiRequest(request); @@ -23,16 +24,35 @@ export async function loader({ request, params }: LoaderFunctionArgs) { return json({ model }); } +// Keep in sync with admin.api.v1.llm-models.ts. +const ModelSourceSchema = z.enum([ + "default", + "admin", + "langfuse", + "auto", + "research", + "provider-api", +]); + const UpdateModelSchema = z.object({ modelName: z.string().min(1).optional(), matchPattern: z.string().min(1).optional(), startDate: z.string().nullable().optional(), + source: ModelSourceSchema.optional(), provider: z.string().nullable().optional(), description: z.string().nullable().optional(), contextWindow: z.number().int().nullable().optional(), maxOutputTokens: z.number().int().nullable().optional(), capabilities: z.array(z.string()).optional(), isHidden: z.boolean().optional(), + needsReview: z.boolean().optional(), + resolvedAt: z.string().datetime().nullable().optional(), + releaseDate: z.string().datetime().nullable().optional(), + deprecationDate: z.string().datetime().nullable().optional(), + knowledgeCutoff: z.string().datetime().nullable().optional(), + supportsStructuredOutput: z.boolean().nullable().optional(), + supportsParallelToolCalls: z.boolean().nullable().optional(), + supportsStreamingToolCalls: z.boolean().nullable().optional(), pricingTiers: z .array( z.object({ @@ -66,6 +86,7 @@ export async function action({ request, params }: ActionFunctionArgs) { } await prisma.llmModel.delete({ where: { id: modelId } }); + await publishLlmRegistryReload(`admin-delete:${existing.source}`); return json({ success: true }); } @@ -86,7 +107,27 @@ export async function action({ request, params }: ActionFunctionArgs) { return json({ error: "Invalid request body", details: parsed.error.issues }, { status: 400 }); } - const { modelName, matchPattern, startDate, pricingTiers, provider, description, contextWindow, maxOutputTokens, capabilities, isHidden } = parsed.data; + const { + modelName, + matchPattern, + startDate, + source, + pricingTiers, + provider, + description, + contextWindow, + maxOutputTokens, + capabilities, + isHidden, + needsReview, + resolvedAt, + releaseDate, + deprecationDate, + knowledgeCutoff, + supportsStructuredOutput, + supportsParallelToolCalls, + supportsStreamingToolCalls, + } = parsed.data; // Validate regex if provided — strip (?i) POSIX flag since our registry handles it if (matchPattern) { @@ -106,12 +147,29 @@ export async function action({ request, params }: ActionFunctionArgs) { ...(modelName !== undefined && { modelName }), ...(matchPattern !== undefined && { matchPattern }), ...(startDate !== undefined && { startDate: startDate ? new Date(startDate) : null }), + ...(source !== undefined && { source }), ...(provider !== undefined && { provider }), ...(description !== undefined && { description }), ...(contextWindow !== undefined && { contextWindow }), ...(maxOutputTokens !== undefined && { maxOutputTokens }), ...(capabilities !== undefined && { capabilities }), ...(isHidden !== undefined && { isHidden }), + ...(needsReview !== undefined && { needsReview }), + ...(resolvedAt !== undefined && { + resolvedAt: resolvedAt ? new Date(resolvedAt) : null, + }), + ...(releaseDate !== undefined && { + releaseDate: releaseDate ? new Date(releaseDate) : null, + }), + ...(deprecationDate !== undefined && { + deprecationDate: deprecationDate ? new Date(deprecationDate) : null, + }), + ...(knowledgeCutoff !== undefined && { + knowledgeCutoff: knowledgeCutoff ? new Date(knowledgeCutoff) : null, + }), + ...(supportsStructuredOutput !== undefined && { supportsStructuredOutput }), + ...(supportsParallelToolCalls !== undefined && { supportsParallelToolCalls }), + ...(supportsStreamingToolCalls !== undefined && { supportsStreamingToolCalls }), }, }); @@ -146,5 +204,7 @@ export async function action({ request, params }: ActionFunctionArgs) { }); }); + await publishLlmRegistryReload(`admin-update:${updated?.source ?? "unknown"}`); + return json({ model: updated }); } diff --git a/apps/webapp/app/routes/admin.api.v1.llm-models.reload.ts b/apps/webapp/app/routes/admin.api.v1.llm-models.reload.ts index 26eee6d84..4864b7fae 100644 --- a/apps/webapp/app/routes/admin.api.v1.llm-models.reload.ts +++ b/apps/webapp/app/routes/admin.api.v1.llm-models.reload.ts @@ -1,6 +1,9 @@ import { type ActionFunctionArgs, json } from "@remix-run/server-runtime"; import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; -import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server"; +import { + llmPricingRegistry, + publishLlmRegistryReload, +} from "~/v3/llmPricingRegistry.server"; export async function action({ request }: ActionFunctionArgs) { await requireAdminApiRequest(request); @@ -9,7 +12,14 @@ export async function action({ request }: ActionFunctionArgs) { return json({ error: "LLM cost tracking is disabled" }, { status: 400 }); } + // Reload this replica immediately so the admin UI sees changes instantly … await llmPricingRegistry.reload(); + // … and notify the other webapp replicas via pub/sub so their in-memory + // registries catch up within seconds. This is the endpoint the cloud-repo + // trigger.dev tasks call after upserting rows. + const url = new URL(request.url); + const reason = url.searchParams.get("reason") ?? "admin-reload"; + await publishLlmRegistryReload(reason); return json({ success: true, message: "LLM pricing registry reloaded" }); } diff --git a/apps/webapp/app/routes/admin.api.v1.llm-models.seed.ts b/apps/webapp/app/routes/admin.api.v1.llm-models.seed.ts index ef85d4587..bccee6732 100644 --- a/apps/webapp/app/routes/admin.api.v1.llm-models.seed.ts +++ b/apps/webapp/app/routes/admin.api.v1.llm-models.seed.ts @@ -2,7 +2,10 @@ import { type ActionFunctionArgs, json } from "@remix-run/server-runtime"; import { seedLlmPricing, syncLlmCatalog } from "@internal/llm-model-catalog"; import { prisma } from "~/db.server"; import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; -import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server"; +import { + llmPricingRegistry, + publishLlmRegistryReload, +} from "~/v3/llmPricingRegistry.server"; export async function action({ request }: ActionFunctionArgs) { await requireAdminApiRequest(request); @@ -16,6 +19,7 @@ export async function action({ request }: ActionFunctionArgs) { if (llmPricingRegistry) { await llmPricingRegistry.reload(); } + await publishLlmRegistryReload("admin-sync"); return json({ success: true, @@ -30,6 +34,7 @@ export async function action({ request }: ActionFunctionArgs) { if (llmPricingRegistry) { await llmPricingRegistry.reload(); } + await publishLlmRegistryReload("admin-seed"); return json({ success: true, diff --git a/apps/webapp/app/routes/admin.api.v1.llm-models.ts b/apps/webapp/app/routes/admin.api.v1.llm-models.ts index 7c8a3d8ca..6427c92eb 100644 --- a/apps/webapp/app/routes/admin.api.v1.llm-models.ts +++ b/apps/webapp/app/routes/admin.api.v1.llm-models.ts @@ -3,6 +3,7 @@ import { z } from "zod"; import { prisma } from "~/db.server"; import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; import { generateFriendlyId } from "~/v3/friendlyIdentifiers"; +import { publishLlmRegistryReload } from "~/v3/llmPricingRegistry.server"; export async function loader({ request }: LoaderFunctionArgs) { await requireAdminApiRequest(request); @@ -30,17 +31,40 @@ export async function loader({ request }: LoaderFunctionArgs) { return json({ models, total, page, pageSize }); } +// Widened to accept the new upstream sources produced by the billing-app +// trigger.dev tasks. "langfuse" for Langfuse-synced rows, "auto" for rows +// auto-priced from provider-reported span cost, "research" for rows upserted +// by the Claude-driven research task, "provider-api" for rows polled from a +// provider's /v1/models endpoint. Keep in sync with the `source` column in +// prisma/schema.prisma. +const ModelSourceSchema = z.enum([ + "default", + "admin", + "langfuse", + "auto", + "research", + "provider-api", +]); + const CreateModelSchema = z.object({ modelName: z.string().min(1), matchPattern: z.string().min(1), startDate: z.string().optional(), - source: z.enum(["default", "admin"]).optional().default("admin"), + source: ModelSourceSchema.optional().default("admin"), provider: z.string().optional(), description: z.string().optional(), contextWindow: z.number().int().optional(), maxOutputTokens: z.number().int().optional(), capabilities: z.array(z.string()).optional(), isHidden: z.boolean().optional(), + needsReview: z.boolean().optional(), + resolvedAt: z.string().datetime().optional(), + releaseDate: z.string().datetime().optional(), + deprecationDate: z.string().datetime().optional(), + knowledgeCutoff: z.string().datetime().optional(), + supportsStructuredOutput: z.boolean().optional(), + supportsParallelToolCalls: z.boolean().optional(), + supportsStreamingToolCalls: z.boolean().optional(), pricingTiers: z.array( z.object({ name: z.string().min(1), @@ -80,7 +104,27 @@ export async function action({ request }: ActionFunctionArgs) { return json({ error: "Invalid request body", details: parsed.error.issues }, { status: 400 }); } - const { modelName, matchPattern, startDate, source, pricingTiers, provider, description, contextWindow, maxOutputTokens, capabilities, isHidden } = parsed.data; + const { + modelName, + matchPattern, + startDate, + source, + pricingTiers, + provider, + description, + contextWindow, + maxOutputTokens, + capabilities, + isHidden, + needsReview, + resolvedAt, + releaseDate, + deprecationDate, + knowledgeCutoff, + supportsStructuredOutput, + supportsParallelToolCalls, + supportsStreamingToolCalls, + } = parsed.data; // Validate regex pattern — strip (?i) POSIX flag since our registry handles it try { @@ -105,6 +149,14 @@ export async function action({ request }: ActionFunctionArgs) { maxOutputTokens: maxOutputTokens ?? null, capabilities: capabilities ?? [], isHidden: isHidden ?? false, + needsReview: needsReview ?? false, + resolvedAt: resolvedAt ? new Date(resolvedAt) : null, + releaseDate: releaseDate ? new Date(releaseDate) : null, + deprecationDate: deprecationDate ? new Date(deprecationDate) : null, + knowledgeCutoff: knowledgeCutoff ? new Date(knowledgeCutoff) : null, + supportsStructuredOutput: supportsStructuredOutput ?? null, + supportsParallelToolCalls: supportsParallelToolCalls ?? null, + supportsStreamingToolCalls: supportsStreamingToolCalls ?? null, }, }); @@ -135,5 +187,11 @@ export async function action({ request }: ActionFunctionArgs) { }); }); + // Notify all webapp replicas to reload the in-memory pricing registry. Rows + // with needsReview=true are excluded from the registry anyway, so we could + // skip the publish for those, but a reload is cheap and keeps the code path + // uniform. + await publishLlmRegistryReload(`admin-create:${source}`); + return json({ model: created }, { status: 201 }); } diff --git a/apps/webapp/app/routes/admin.llm-models.$modelId.tsx b/apps/webapp/app/routes/admin.llm-models.$modelId.tsx index 7b51067dd..52fe436d2 100644 --- a/apps/webapp/app/routes/admin.llm-models.$modelId.tsx +++ b/apps/webapp/app/routes/admin.llm-models.$modelId.tsx @@ -9,7 +9,10 @@ import { Input } from "~/components/primitives/Input"; import { Paragraph } from "~/components/primitives/Paragraph"; import { prisma } from "~/db.server"; import { requireUserId } from "~/services/session.server"; -import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server"; +import { + llmPricingRegistry, + publishLlmRegistryReload, +} from "~/v3/llmPricingRegistry.server"; export const loader = async ({ request, params }: LoaderFunctionArgs) => { const userId = await requireUserId(request); @@ -65,6 +68,7 @@ export async function action({ request, params }: ActionFunctionArgs) { if (_action === "delete") { await prisma.llmModel.delete({ where: { id: modelId } }); await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-delete"); return redirect("/admin/llm-models"); } @@ -138,6 +142,7 @@ export async function action({ request, params }: ActionFunctionArgs) { } await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-update"); return typedjson({ success: true }); } diff --git a/apps/webapp/app/routes/admin.llm-models._index.tsx b/apps/webapp/app/routes/admin.llm-models._index.tsx index ea2eff725..583f98729 100644 --- a/apps/webapp/app/routes/admin.llm-models._index.tsx +++ b/apps/webapp/app/routes/admin.llm-models._index.tsx @@ -21,7 +21,10 @@ import { prisma } from "~/db.server"; import { requireUserId } from "~/services/session.server"; import { createSearchParams } from "~/utils/searchParams"; import { seedLlmPricing, syncLlmCatalog } from "@internal/llm-model-catalog"; -import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server"; +import { + llmPricingRegistry, + publishLlmRegistryReload, +} from "~/v3/llmPricingRegistry.server"; const PAGE_SIZE = 50; @@ -89,6 +92,7 @@ export async function action({ request }: ActionFunctionArgs) { const result = await seedLlmPricing(prisma); console.log(`[admin] seed complete: ${result.modelsCreated} created, ${result.modelsSkipped} skipped, ${result.modelsUpdated} updated`); await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-seed"); console.log("[admin] registry reloaded after seed"); return typedjson({ success: true, @@ -101,6 +105,7 @@ export async function action({ request }: ActionFunctionArgs) { const result = await syncLlmCatalog(prisma); console.log(`[admin] sync complete: ${result.modelsUpdated} updated, ${result.modelsSkipped} skipped`); await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-sync"); console.log("[admin] registry reloaded after sync"); return typedjson({ success: true, @@ -111,6 +116,7 @@ export async function action({ request }: ActionFunctionArgs) { if (_action === "reload") { console.log("[admin] reload action started"); await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-reload"); console.log("[admin] registry reloaded"); return typedjson({ success: true, message: "Registry reloaded" }); } @@ -139,6 +145,7 @@ export async function action({ request }: ActionFunctionArgs) { if (typeof modelId === "string") { await prisma.llmModel.delete({ where: { id: modelId } }); await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-delete"); } return typedjson({ success: true }); } diff --git a/apps/webapp/app/routes/admin.llm-models.new.tsx b/apps/webapp/app/routes/admin.llm-models.new.tsx index 7f18bf582..b593c988a 100644 --- a/apps/webapp/app/routes/admin.llm-models.new.tsx +++ b/apps/webapp/app/routes/admin.llm-models.new.tsx @@ -9,7 +9,10 @@ import { Input } from "~/components/primitives/Input"; import { prisma } from "~/db.server"; import { requireUserId } from "~/services/session.server"; import { generateFriendlyId } from "~/v3/friendlyIdentifiers"; -import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server"; +import { + llmPricingRegistry, + publishLlmRegistryReload, +} from "~/v3/llmPricingRegistry.server"; export const loader = async ({ request }: LoaderFunctionArgs) => { const userId = await requireUserId(request); @@ -105,6 +108,7 @@ export async function action({ request }: ActionFunctionArgs) { } await llmPricingRegistry?.reload(); + await publishLlmRegistryReload("admin-ui-create"); return redirect(`/admin/llm-models/${model.friendlyId}`); } diff --git a/apps/webapp/app/v3/llmPricingRegistry.server.ts b/apps/webapp/app/v3/llmPricingRegistry.server.ts index 2212c4177..5b8a37b0b 100644 --- a/apps/webapp/app/v3/llmPricingRegistry.server.ts +++ b/apps/webapp/app/v3/llmPricingRegistry.server.ts @@ -1,8 +1,10 @@ import { ModelPricingRegistry, seedLlmPricing } from "@internal/llm-model-catalog"; import { prisma, $replica } from "~/db.server"; import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; import { signalsEmitter } from "~/services/signals.server"; import { singleton } from "~/utils/singleton"; +import { llmRegistryPubSub } from "./services/llmRegistryPubSub.server"; import { setLlmPricingRegistry } from "./utils/enrichCreatableEvents.server"; async function initRegistry(registry: ModelPricingRegistry) { @@ -27,7 +29,10 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => { console.error("Failed to initialize LLM pricing registry", err); }); - // Periodic reload + // Periodic reload — acts as a backstop in case a pub/sub message is missed + // (e.g. during a Redis failover). The primary reload path is the pub/sub + // subscriber below which reacts within seconds of an admin mutation or a + // trigger.dev task upserting rows. const reloadInterval = env.LLM_PRICING_RELOAD_INTERVAL_MS; const interval = setInterval(() => { registry.reload().catch((err) => { @@ -35,16 +40,43 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => { }); }, reloadInterval); - signalsEmitter.on("SIGTERM", () => { + // Realtime reloads across webapp replicas. Publishers include admin routes + // and the billing-app trigger.dev LLM registry tasks (via the + // /admin/api/v1/llm-models/reload endpoint). + let unsubscribe: (() => Promise) | undefined; + llmRegistryPubSub + .subscribe(async (reason) => { + logger.info("Reloading LLM pricing registry from pub/sub", { reason }); + await registry.reload(); + }) + .then((fn) => { + unsubscribe = fn; + }) + .catch((err) => { + logger.error("Failed to subscribe to llm-registry reload channel", { error: err }); + }); + + const cleanup = () => { clearInterval(interval); - }); - signalsEmitter.on("SIGINT", () => { - clearInterval(interval); - }); + unsubscribe?.().catch(() => undefined); + }; + + signalsEmitter.on("SIGTERM", cleanup); + signalsEmitter.on("SIGINT", cleanup); return registry; }); +/** + * Publish a reload event so every webapp replica reloads its registry. Safe to + * call after any admin mutation; the pub/sub message is tiny and subscribers + * deduplicate via the in-memory patterns list. + */ +export async function publishLlmRegistryReload(reason: string): Promise { + if (!env.LLM_COST_TRACKING_ENABLED) return; + await llmRegistryPubSub.publishReload(reason); +} + /** * Wait for the LLM pricing registry to finish its initial load, with a timeout. * After the first call resolves (or times out), subsequent calls are no-ops. diff --git a/apps/webapp/app/v3/services/llmRegistryPubSub.server.ts b/apps/webapp/app/v3/services/llmRegistryPubSub.server.ts new file mode 100644 index 000000000..8d2fd7313 --- /dev/null +++ b/apps/webapp/app/v3/services/llmRegistryPubSub.server.ts @@ -0,0 +1,109 @@ +import { EventEmitter } from "node:events"; +import { env } from "~/env.server"; +import { createRedisClient, type RedisClient, type RedisWithClusterOptions } from "~/redis.server"; +import { logger } from "~/services/logger.server"; +import { singleton } from "~/utils/singleton"; + +/** + * Pub/sub channel used to notify webapp replicas that the in-memory LLM + * pricing registry should reload from Postgres. Published by admin routes + * whenever an LlmModel row is mutated, and by the billing-app trigger.dev + * tasks (via the admin reload endpoint) whenever they upsert rows. + */ +export const LLM_REGISTRY_RELOAD_CHANNEL = "llm-registry:reload"; + +type LlmRegistryPubSubOptions = { + redis: RedisWithClusterOptions; +}; + +/** + * Thin publish/subscribe wrapper around ioredis for the LLM pricing registry. + * The payload is intentionally tiny (a timestamp) because subscribers always + * perform a full reload — there's no incremental diff. + */ +export class LlmRegistryPubSub { + private _publisher: RedisClient; + private _subscriber: RedisClient | null = null; + private _emitter = new EventEmitter(); + + constructor(private _options: LlmRegistryPubSubOptions) { + this._publisher = createRedisClient("llm-registry:publisher", this._options.redis); + } + + /** Notifies all webapp replicas that they should reload the registry. */ + async publishReload(reason?: string): Promise { + try { + await this._publisher.publish( + LLM_REGISTRY_RELOAD_CHANNEL, + JSON.stringify({ at: new Date().toISOString(), reason: reason ?? "unspecified" }) + ); + } catch (error) { + logger.error("Failed to publish llm-registry reload", { error }); + } + } + + /** + * Subscribes this process to reload notifications. The supplied handler is + * called every time any replica publishes a reload; it should be idempotent + * and reasonably fast (the caller should usually just invoke + * `registry.reload()`). + */ + async subscribe(handler: (reason: string) => void | Promise): Promise<() => Promise> { + if (this._subscriber) { + throw new Error("LlmRegistryPubSub already subscribed from this process"); + } + + const subscriber = createRedisClient("llm-registry:subscriber", this._options.redis); + this._subscriber = subscriber; + + await subscriber.subscribe(LLM_REGISTRY_RELOAD_CHANNEL); + + const messageHandler = (channel: string, message: string) => { + if (channel !== LLM_REGISTRY_RELOAD_CHANNEL) return; + + let reason = "unspecified"; + try { + const parsed = JSON.parse(message) as { reason?: string }; + if (typeof parsed.reason === "string") reason = parsed.reason; + } catch { + // Old-format message — ignore the parse error, the handler fires anyway. + } + + Promise.resolve(handler(reason)).catch((error) => { + logger.error("llm-registry reload handler threw", { error, reason }); + }); + + this._emitter.emit("reload", reason); + }; + + subscriber.on("message", messageHandler); + + return async () => { + subscriber.off("message", messageHandler); + try { + await subscriber.unsubscribe(LLM_REGISTRY_RELOAD_CHANNEL); + } catch { + // Ignore — we're shutting down. + } + await subscriber.quit().catch(() => undefined); + this._subscriber = null; + }; + } + + on(event: "reload", listener: (reason: string) => void): void { + this._emitter.on(event, listener); + } +} + +export const llmRegistryPubSub = singleton("llmRegistryPubSub", () => { + return new LlmRegistryPubSub({ + redis: { + port: env.PUBSUB_REDIS_PORT, + host: env.PUBSUB_REDIS_HOST, + username: env.PUBSUB_REDIS_USERNAME, + password: env.PUBSUB_REDIS_PASSWORD, + tlsDisabled: env.PUBSUB_REDIS_TLS_DISABLED === "true", + clusterMode: env.PUBSUB_REDIS_CLUSTER_MODE_ENABLED === "1", + }, + }); +}); diff --git a/apps/webapp/app/v3/utils/enrichCreatableEvents.server.ts b/apps/webapp/app/v3/utils/enrichCreatableEvents.server.ts index 643820104..b2647087b 100644 --- a/apps/webapp/app/v3/utils/enrichCreatableEvents.server.ts +++ b/apps/webapp/app/v3/utils/enrichCreatableEvents.server.ts @@ -1,4 +1,5 @@ import { modelCatalog } from "@internal/llm-model-catalog"; +import { metrics } from "@opentelemetry/api"; import type { CreateEventInput, LlmMetricsData } from "../eventRepository/eventRepository.types"; // Registry interface — matches ModelPricingRegistry from @internal/llm-model-catalog @@ -23,6 +24,41 @@ let _registry: CostRegistry | undefined; const ENRICHABLE_KINDS = new Set(["INTERNAL", "SERVER", "CLIENT", "CONSUMER", "PRODUCER"]); +// Low-cardinality allowlist of gen_ai.system values. Anything outside this set +// is collapsed into "other" to keep the metric cardinality bounded. Keep in +// sync with the OpenTelemetry GenAI semantic conventions. +const KNOWN_GEN_AI_SYSTEMS = new Set([ + "openai", + "anthropic", + "google", + "vertex_ai", + "aws.bedrock", + "az.ai.openai", + "az.ai.inference", + "cohere", + "deepseek", + "groq", + "ibm.watsonx.ai", + "mistral_ai", + "perplexity", + "xai", +]); + +function normalizeGenAiSystem(value: unknown): string { + if (typeof value !== "string" || value.length === 0) return "unknown"; + const normalized = value.toLowerCase(); + return KNOWN_GEN_AI_SYSTEMS.has(normalized) ? normalized : "other"; +} + +// Emits a metric whenever we see an LLM span with usage data but can't resolve +// a price from the in-memory registry. Used by the llm-registry productionization +// pipeline to detect missing models without having to query ClickHouse. +const llmMeter = metrics.getMeter("trigger.dev.llm_registry", "1.0.0"); +const missingModelCounter = llmMeter.createCounter("llm_missing_model_enrichment", { + description: + "LLM spans with gen_ai.response.model + usage data that could not be priced by the registry", +}); + export function setLlmPricingRegistry(registry: CostRegistry): void { _registry = registry; } @@ -103,6 +139,17 @@ function enrichLlmMetrics(event: CreateEventInput): void { providerCost = extractProviderCost(props); } + // Observability: if the registry is loaded but couldn't price this span, emit + // a low-cardinality counter so the productionization pipeline can detect + // missing models without hitting ClickHouse. We tag by gen_ai.system (bounded + // enum) and whether a provider-reported cost rescued the span. + if (_registry?.isLoaded && !cost) { + missingModelCounter.add(1, { + gen_ai_system: normalizeGenAiSystem(props["gen_ai.system"]), + has_provider_cost: providerCost !== null, + }); + } + if (cost) { // Add trigger.llm.* attributes to the span from our pricing registry event.properties = { diff --git a/internal-packages/database/prisma/migrations/20260417233744_add_llm_model_catalog_metadata/migration.sql b/internal-packages/database/prisma/migrations/20260417233744_add_llm_model_catalog_metadata/migration.sql new file mode 100644 index 000000000..eef67eb86 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260417233744_add_llm_model_catalog_metadata/migration.sql @@ -0,0 +1,9 @@ +-- AlterTable +ALTER TABLE "public"."llm_models" ADD COLUMN "deprecation_date" TIMESTAMP(3), +ADD COLUMN "knowledge_cutoff" TIMESTAMP(3), +ADD COLUMN "needs_review" BOOLEAN NOT NULL DEFAULT false, +ADD COLUMN "release_date" TIMESTAMP(3), +ADD COLUMN "resolved_at" TIMESTAMP(3), +ADD COLUMN "supports_parallel_tool_calls" BOOLEAN, +ADD COLUMN "supports_streaming_tool_calls" BOOLEAN, +ADD COLUMN "supports_structured_output" BOOLEAN; diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index 1462752d8..7c4ee640c 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -2694,18 +2694,40 @@ model LlmModel { modelName String @map("model_name") matchPattern String @map("match_pattern") startDate DateTime? @map("start_date") - source String @default("default") // "default", "admin", "project" + /// Where the row originated. Known values: "default", "admin", "project", + /// "langfuse" (synced from the Langfuse model-prices JSON), + /// "auto" (priced from provider-reported span cost), "research" (upserted + /// by the Claude-driven research task), "provider-api" (polled from a + /// provider's models endpoint). + source String @default("default") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") // Catalog metadata for model registry - provider String? @map("provider") - description String? @map("description") - contextWindow Int? @map("context_window") - maxOutputTokens Int? @map("max_output_tokens") - capabilities String[] @default([]) @map("capabilities") - isHidden Boolean @default(false) @map("is_hidden") - baseModelName String? @map("base_model_name") + provider String? @map("provider") + description String? @map("description") + contextWindow Int? @map("context_window") + maxOutputTokens Int? @map("max_output_tokens") + capabilities String[] @default([]) @map("capabilities") + isHidden Boolean @default(false) @map("is_hidden") + baseModelName String? @map("base_model_name") + /// When this model row was last resolved by the research task. Used by the + /// refresh-stale job to re-research rows that are older than the staleness + /// window (default 7 days). + resolvedAt DateTime? @map("resolved_at") + /// When true, the row is hidden from the in-memory pricing registry until a + /// human approves it. Used to gate auto-priced and freshly-synced rows. + needsReview Boolean @default(false) @map("needs_review") + /// When the model was announced / released. Populated by the research task. + releaseDate DateTime? @map("release_date") + /// When the model was (or will be) deprecated. Populated by the research task. + deprecationDate DateTime? @map("deprecation_date") + /// Knowledge cutoff of the model's training data. + knowledgeCutoff DateTime? @map("knowledge_cutoff") + /// Capability flags populated by the research task. Null means unknown. + supportsStructuredOutput Boolean? @map("supports_structured_output") + supportsParallelToolCalls Boolean? @map("supports_parallel_tool_calls") + supportsStreamingToolCalls Boolean? @map("supports_streaming_tool_calls") pricingTiers LlmPricingTier[] prices LlmPrice[] diff --git a/internal-packages/llm-model-catalog/src/registry.ts b/internal-packages/llm-model-catalog/src/registry.ts index 80da40ba9..ca23c2f19 100644 --- a/internal-packages/llm-model-catalog/src/registry.ts +++ b/internal-packages/llm-model-catalog/src/registry.ts @@ -45,7 +45,14 @@ export class ModelPricingRegistry { async loadFromDatabase(): Promise { const models = await this._prisma.llmModel.findMany({ - where: { projectId: null }, + where: { + projectId: null, + // Exclude rows awaiting admin approval (e.g. auto-priced rows written + // by the detect-missing-models trigger.dev task, or freshly-synced + // Langfuse rows that haven't been reviewed yet). These fields were + // added as part of the llm-registry productionization work. + needsReview: false, + }, include: { pricingTiers: { include: { prices: true },