feat: enrich assistant request telemetry

2026-08-02 08:51:48 +02:00
1 parent d562706c8b
commit 04520cf67c
@@ -0,0 +1,140 @@
+import { fetchWithTrustedCertificatesAsync } from "@homarr/core/infrastructure/http";
+
+export type OpenRouterGenerationTelemetry = {
+  generationId?: string;
+  routedProvider?: string;
+  providerLatencyMs?: number;
+  generationTimeMs?: number;
+  moderationLatencyMs?: number;
+  providerOutputTokensPerSecond?: number;
+  inputTokens?: number;
+  outputTokens?: number;
+  normalizedInputTokens?: number;
+  normalizedOutputTokens?: number;
+  cachedInputTokens?: number;
+  reasoningTokens?: number;
+  cost?: number;
+  upstreamCost?: number;
+  cacheDiscount?: number;
+  finishReason?: string;
+  nativeFinishReason?: string;
+  serviceTier?: string;
+  dataRegion?: string;
+  fallbackCount?: number;
+  fallbackLatencyMs?: number;
+  isByok?: boolean;
+  streamed?: boolean;
+  cancelled?: boolean;
+};
+
+const asRecord = (value: unknown): Record<string, unknown> | null =>
+  typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record<string, unknown>) : null;
+
+const asFiniteNonNegativeNumber = (value: unknown) => {
+  const number = typeof value === "number" ? value : typeof value === "string" ? Number(value) : Number.NaN;
+  return Number.isFinite(number) && number >= 0 ? number : undefined;
+};
+
+const asString = (value: unknown) => (typeof value === "string" && value.length > 0 ? value : undefined);
+const asBoolean = (value: unknown) => (typeof value === "boolean" ? value : undefined);
+
+export const parseOpenRouterGenerationTelemetry = (value: unknown): OpenRouterGenerationTelemetry | null => {
+  const root = asRecord(value);
+  const data = asRecord(root?.data) ?? root;
+  if (!data) return null;
+
+  const normalizedInputTokens = asFiniteNonNegativeNumber(data.tokens_prompt);
+  const normalizedOutputTokens = asFiniteNonNegativeNumber(data.tokens_completion);
+  const inputTokens = asFiniteNonNegativeNumber(data.native_tokens_prompt) ?? normalizedInputTokens;
+  const outputTokens = asFiniteNonNegativeNumber(data.native_tokens_completion) ?? normalizedOutputTokens;
+  const generationTimeMs = asFiniteNonNegativeNumber(data.generation_time);
+  const providerResponses = Array.isArray(data.provider_responses) ? data.provider_responses : undefined;
+  const parsedProviderResponses = providerResponses?.map(asRecord).filter((response) => response !== null);
+  const failedProviderResponses = parsedProviderResponses?.filter((response) => {
+    const status = asFiniteNonNegativeNumber(response.status);
+    return status !== undefined && (status < 200 || status >= 300);
+  });
+  const fallbackCount = providerResponses
+    ? failedProviderResponses && failedProviderResponses.length > 0
+      ? failedProviderResponses.length
+      : Math.max(providerResponses.length - 1, 0)
+    : undefined;
+  const fallbackLatencies = failedProviderResponses?.map((response) => asFiniteNonNegativeNumber(response.latency));
+  const fallbackLatencyMs =
+    fallbackLatencies && fallbackLatencies.length > 0 && fallbackLatencies.every((latency) => latency !== undefined)
+      ? (fallbackLatencies as number[]).reduce((sum, latency) => sum + latency, 0)
+      : undefined;
+  const providerOutputTokensPerSecond =
+    outputTokens !== undefined && generationTimeMs !== undefined && generationTimeMs > 0
+      ? outputTokens / (generationTimeMs / 1000)
+      : undefined;
+  const generationId = asString(data.id);
+  const routedProvider = asString(data.provider_name);
+  const finishReason = asString(data.finish_reason);
+  const nativeFinishReason = asString(data.native_finish_reason);
+  const serviceTier = asString(data.service_tier);
+  const dataRegion = asString(data.data_region);
+
+  const telemetry: OpenRouterGenerationTelemetry = {
+    ...(generationId ? { generationId } : {}),
+    ...(routedProvider ? { routedProvider } : {}),
+    ...(asFiniteNonNegativeNumber(data.latency) !== undefined
+      ? { providerLatencyMs: asFiniteNonNegativeNumber(data.latency) }
+      : {}),
+    ...(generationTimeMs !== undefined ? { generationTimeMs } : {}),
+    ...(asFiniteNonNegativeNumber(data.moderation_latency) !== undefined
+      ? { moderationLatencyMs: asFiniteNonNegativeNumber(data.moderation_latency) }
+      : {}),
+    ...(providerOutputTokensPerSecond !== undefined ? { providerOutputTokensPerSecond } : {}),
+    ...(inputTokens !== undefined ? { inputTokens } : {}),
+    ...(outputTokens !== undefined ? { outputTokens } : {}),
+    ...(normalizedInputTokens !== undefined ? { normalizedInputTokens } : {}),
+    ...(normalizedOutputTokens !== undefined ? { normalizedOutputTokens } : {}),
+    ...(asFiniteNonNegativeNumber(data.native_tokens_cached) !== undefined
+      ? { cachedInputTokens: asFiniteNonNegativeNumber(data.native_tokens_cached) }
+      : {}),
+    ...(asFiniteNonNegativeNumber(data.native_tokens_reasoning) !== undefined
+      ? { reasoningTokens: asFiniteNonNegativeNumber(data.native_tokens_reasoning) }
+      : {}),
+    ...(asFiniteNonNegativeNumber(data.total_cost) !== undefined
+      ? { cost: asFiniteNonNegativeNumber(data.total_cost) }
+      : {}),
+    ...(asFiniteNonNegativeNumber(data.upstream_inference_cost) !== undefined
+      ? { upstreamCost: asFiniteNonNegativeNumber(data.upstream_inference_cost) }
+      : {}),
+    ...(asFiniteNonNegativeNumber(data.cache_discount) !== undefined
+      ? { cacheDiscount: asFiniteNonNegativeNumber(data.cache_discount) }
+      : {}),
+    ...(finishReason ? { finishReason } : {}),
+    ...(nativeFinishReason ? { nativeFinishReason } : {}),
+    ...(serviceTier ? { serviceTier } : {}),
+    ...(dataRegion ? { dataRegion } : {}),
+    ...(fallbackCount !== undefined ? { fallbackCount } : {}),
+    ...(fallbackLatencyMs !== undefined ? { fallbackLatencyMs } : {}),
+    ...(asBoolean(data.is_byok) !== undefined ? { isByok: asBoolean(data.is_byok) } : {}),
+    ...(asBoolean(data.streamed) !== undefined ? { streamed: asBoolean(data.streamed) } : {}),
+    ...(asBoolean(data.cancelled) !== undefined ? { cancelled: asBoolean(data.cancelled) } : {}),
+  };
+
+  return Object.keys(telemetry).length > 0 ? telemetry : null;
+};
+
+export const fetchOpenRouterGenerationTelemetryAsync = async ({
+  baseUrl,
+  generationId,
+  headers,
+}: {
+  baseUrl: string;
+  generationId: string;
+  headers: Record<string, string>;
+}): Promise<OpenRouterGenerationTelemetry | null> => {
+  try {
+    const endpoint = new URL(`${baseUrl.replace(/\/$/, "")}/generation`);
+    endpoint.searchParams.set("id", generationId);
+    const response = await fetchWithTrustedCertificatesAsync(endpoint, { headers, timeout: 3000 });
+    if (!response.ok) return null;
+    return parseOpenRouterGenerationTelemetry(await response.json());
+  } catch {
+    return null;
+  }
+};
@@ -4,3 +4,4 @@ export {
  getSelectedModelDetailsAsync,
  type AssistantContextEntity,
 } from "./router/assistant";
+export type { OpenRouterGenerationTelemetry } from "./assistant-generation-telemetry";
@@ -25,8 +25,9 @@ import {
  resolveAssistantModelId,
 } from "@homarr/definitions";

-import { createTRPCRouter, permissionRequiredProcedure, protectedProcedure } from "../trpc";
 import type { createTRPCContext } from "../trpc";
+import { fetchOpenRouterGenerationTelemetryAsync } from "../assistant-generation-telemetry";
+import { createTRPCRouter, permissionRequiredProcedure, protectedProcedure } from "../trpc";
 import { boardRouter } from "./board";

 const adminProcedure = permissionRequiredProcedure.requiresPermission("admin");
@@ -159,6 +160,25 @@ const getProviderHeaders = (configuration: AssistantConfiguration) => {
  return { ...headers, ...decryptCustomHeaders(configuration.encryptedHeaders) };
 };

+const getGenerationIdsFromMessageContent = (content: unknown) => {
+  if (!content || typeof content !== "object" || Array.isArray(content)) return [];
+  const metadata = (content as { metadata?: unknown }).metadata;
+  if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return [];
+  const custom = (metadata as { custom?: unknown }).custom;
+  if (!custom || typeof custom !== "object" || Array.isArray(custom)) return [];
+  const telemetry = (custom as { telemetry?: unknown }).telemetry;
+  if (!telemetry || typeof telemetry !== "object" || Array.isArray(telemetry)) return [];
+  const steps = (telemetry as { steps?: unknown }).steps;
+  if (!Array.isArray(steps)) return [];
+  return steps
+    .flatMap((step) => {
+      if (!step || typeof step !== "object" || Array.isArray(step)) return [];
+      const generationId = (step as { generationId?: unknown }).generationId;
+      return typeof generationId === "string" && /^gen-[A-Za-z0-9_-]{1,128}$/.test(generationId) ? [generationId] : [];
+    })
+    .slice(0, 8);
+};
+
 const fetchModelsAsync = async (configuration: AssistantConfiguration) => {
  if (!configuration.modelDiscoveryPath) {
    throw new TRPCError({
@@ -674,6 +694,43 @@ export const assistantRouter = createTRPCRouter({
    };
  }),

+  getGenerationTelemetry: protectedProcedure
+    .input(
+      z.object({
+        threadId: z.string().max(64),
+        messageId: z.string().min(1).max(128),
+      }),
+    )
+    .query(async ({ ctx, input }) => {
+      const thread = await ownedThreadAsync(ctx.db, input.threadId, ctx.session.user.id);
+      const [message, configuration] = await Promise.all([
+        ctx.db.query.assistantMessages.findFirst({
+          where: and(eq(assistantMessages.id, input.messageId), eq(assistantMessages.threadId, thread.id)),
+        }),
+        getConfigurationAsync(ctx.db),
+      ]);
+      if (!message) throw new TRPCError({ code: "NOT_FOUND", message: "Message not found." });
+      if (configuration?.provider !== "openrouter" || !configuration.encryptedApiKey) {
+        return { complete: true, generations: [] };
+      }
+
+      const generationIds = getGenerationIdsFromMessageContent(parse(message.content));
+      if (generationIds.length === 0) return { complete: true, generations: [] };
+      const generations = await Promise.all(
+        generationIds.map((generationId) =>
+          fetchOpenRouterGenerationTelemetryAsync({
+            baseUrl: configuration.baseUrl,
+            generationId,
+            headers: getProviderHeaders(configuration),
+          }),
+        ),
+      );
+      return {
+        complete: generations.every((generation) => generation !== null),
+        generations: generations.filter((generation) => generation !== null),
+      };
+    }),
+
  renameThread: protectedProcedure
    .input(z.object({ threadId: z.string().max(64), title: z.string().trim().min(1).max(120) }))
    .mutation(async ({ ctx, input }) => {
@@ -0,0 +1,66 @@
+import { describe, expect, test, vi } from "vitest";
+
+vi.mock("@homarr/core/infrastructure/http", () => ({ fetchWithTrustedCertificatesAsync: vi.fn() }));
+
+import { parseOpenRouterGenerationTelemetry } from "../assistant-generation-telemetry";
+
+describe("parseOpenRouterGenerationTelemetry", () => {
+  test("uses provider-native tokens for context and throughput", () => {
+    expect(
+      parseOpenRouterGenerationTelemetry({
+        data: {
+          id: "gen-1",
+          provider_name: "Alibaba",
+          latency: 964,
+          generation_time: 1451,
+          moderation_latency: 12,
+          tokens_prompt: 13_343,
+          tokens_completion: 62,
+          native_tokens_prompt: 12_777,
+          native_tokens_completion: 55,
+          native_tokens_cached: 2048,
+          native_tokens_reasoning: 7,
+          total_cost: 0.00041,
+          upstream_inference_cost: 0.00039,
+          cache_discount: 0.00003,
+          finish_reason: "stop",
+          native_finish_reason: "stop",
+          service_tier: "default",
+          data_region: "US",
+          provider_responses: [
+            { status: 429, latency: 620 },
+            { status: 200, latency: 964 },
+          ],
+          is_byok: false,
+          streamed: true,
+          cancelled: false,
+        },
+      }),
+    ).toMatchObject({
+      generationId: "gen-1",
+      routedProvider: "Alibaba",
+      providerLatencyMs: 964,
+      generationTimeMs: 1451,
+      providerOutputTokensPerSecond: 37.90489317711923,
+      inputTokens: 12_777,
+      outputTokens: 55,
+      normalizedInputTokens: 13_343,
+      normalizedOutputTokens: 62,
+      cachedInputTokens: 2048,
+      reasoningTokens: 7,
+      cost: 0.00041,
+      upstreamCost: 0.00039,
+      cacheDiscount: 0.00003,
+      fallbackCount: 1,
+      fallbackLatencyMs: 620,
+      serviceTier: "default",
+      dataRegion: "US",
+    });
+  });
+
+  test("returns null for malformed generation data", () => {
+    expect(parseOpenRouterGenerationTelemetry(null)).toBeNull();
+    expect(parseOpenRouterGenerationTelemetry({ data: [] })).toBeNull();
+    expect(parseOpenRouterGenerationTelemetry({ data: { latency: -1, total_cost: "invalid" } })).toBeNull();
+  });
+});