feat(llm-registry): phase 0/1 — schema, pub/sub, OTEL counter, CI drift check

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 <eallam@icloud.com>
2026-04-17 23:47:07 +00:00
parent 7d7ebdde52
commit 3e491df445
15 changed files with 447 additions and 23 deletions
@@ -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
@@ -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
@@ -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 });
 }
@@ -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" });
 }
@@ -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,
@@ -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 });
 }
@@ -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 });
  }

@@ -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 });
  }
@@ -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}`);
 }

@@ -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<void>) | 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<void> {
+  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.
@@ -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<void> {
+    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<void>): Promise<() => Promise<void>> {
+    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",
+    },
+  });
+});
@@ -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 = {
@@ -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;
@@ -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[]
@@ -45,7 +45,14 @@ export class ModelPricingRegistry {

  async loadFromDatabase(): Promise<void> {
    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 },