diff --git a/AGENTS.md b/AGENTS.md index 0a2ebc0704..3c3078d6b1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,6 +117,13 @@ E2E tests are organized for optimal performance: - `apps/gateway/src/api-individual.e2e.ts` - Contains individual test cases that need isolation - **Concurrent mode**: The main test suite uses `{ concurrent: true }` to enable parallel execution of `.each()` tests +#### Gateway test harness resets shared state per test + +The gateway integration specs use `createGatewayApiTestHarness()` (`apps/gateway/src/test-utils/gateway-api-test-harness.ts`), whose `beforeEach` **deletes and re-seeds all test data before every test** — including the shared organization, which is always re-seeded with `retentionLevel: "retain"`, `plan: "pro"`, and `credits: "100.00"`. Because of this: + +- A test may freely mutate shared org/project state (`retentionLevel`, `credits`, `plan`, project `mode`, etc.) **without restoring it afterward** — the next test's `beforeEach` re-seed guarantees isolation. Do not add manual restore/cleanup for these mutations; it is redundant and inconsistent with the rest of the suite. +- A test that depends on a specific value should either rely on the documented seed default (e.g. `retentionLevel: "retain"`) or set it explicitly at the top of the test — never on a value left behind by a previous test. + ### Database Operations NOTE: these commands can only be run in the root directory of the repository, not in individual app directories. diff --git a/apps/gateway/src/api.spec.ts b/apps/gateway/src/api.spec.ts index 2024969c8d..528b4747a4 100644 --- a/apps/gateway/src/api.spec.ts +++ b/apps/gateway/src/api.spec.ts @@ -904,6 +904,136 @@ describe("api", () => { ); }); + test("/v1/chat/completions strips log payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id-retention-none", + token: "real-token-retention-none", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-id-retention-none", + token: "sk-test-key", + provider: "openai", + organizationId: "org-id", + baseUrl: mockServerUrl, + }); + + const requestId = "retention-none-request-id"; + const res = await app.request("/v1/chat/completions", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-retention-none", + "x-request-id": requestId, + }, + body: JSON.stringify({ + model: "gpt-4o-mini", + messages: [{ role: "user", content: "Hello!" }], + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const logRow = logs.find((log) => log.requestId === requestId); + + expect(logRow).toBeTruthy(); + // Metadata / metering is still recorded for a non-retaining org... + expect(logRow?.usedProvider).toBe("openai"); + expect(logRow?.finishReason).toBe("stop"); + expect(Number(logRow?.promptTokens)).toBeGreaterThan(0); + // ...but the request/response payload never reaches the database because + // the gateway strips it before publishing to the log queue. + expect(logRow?.messages).toBeNull(); + expect(logRow?.content).toBeNull(); + expect(logRow?.reasoningContent).toBeNull(); + }); + + test("/v1/chat/completions retains log payload when retention is enabled", async () => { + // The seeded org defaults to retentionLevel: "retain". + await db.insert(tables.apiKey).values({ + id: "token-id-retention-retain", + token: "real-token-retention-retain", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-id-retention-retain", + token: "sk-test-key", + provider: "openai", + organizationId: "org-id", + baseUrl: mockServerUrl, + }); + + const requestId = "retention-retain-request-id"; + const res = await app.request("/v1/chat/completions", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-retention-retain", + "x-request-id": requestId, + }, + body: JSON.stringify({ + model: "gpt-4o-mini", + messages: [{ role: "user", content: "Hello!" }], + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const logRow = logs.find((log) => log.requestId === requestId); + + expect(logRow).toBeTruthy(); + expect(logRow?.messages).toEqual([{ role: "user", content: "Hello!" }]); + expect(typeof logRow?.content).toBe("string"); + expect((logRow?.content ?? "").length).toBeGreaterThan(0); + }); + + test("/v1/responses is rejected outright when retention is disabled", async () => { + // The Responses API stores conversation state, so it is gated to + // retaining orgs — a non-retaining org is rejected before any request or + // response payload can be processed or logged. + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id-responses-retention-none", + token: "real-token-responses-retention-none", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + const res = await app.request("/v1/responses", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-responses-retention-none", + }, + body: JSON.stringify({ + model: "gpt-4o-mini", + input: "secret retention payload", + }), + }); + + expect(res.status).toBe(400); + const json = await res.json(); + expect(json.error.code).toBe("data_retention_required"); + }); + test("/v1/chat/completions rejects Vertex service tiers outside the global endpoint", async () => { const originalVertexRegion = process.env.LLM_GOOGLE_VERTEX_REGION; process.env.LLM_GOOGLE_VERTEX_REGION = "us-central1"; @@ -1562,6 +1692,55 @@ describe("api", () => { } }); + test("/v1/moderations does not persist payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id", + token: "real-token", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-id", + token: "sk-test-key", + provider: "openai", + organizationId: "org-id", + baseUrl: mockServerUrl, + }); + + const requestId = "moderation-retention-none-request-id"; + const res = await app.request("/v1/moderations", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token", + "x-request-id": requestId, + }, + body: JSON.stringify({ + input: "I want to attack someone.", + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const moderationLog = logs.find((log) => log.requestId === requestId); + + expect(moderationLog).toBeTruthy(); + expect(moderationLog?.usedProvider).toBe("openai"); + expect(moderationLog?.finishReason).toBe("stop"); + // Payload never reaches the database for a non-retaining org. + expect(moderationLog?.messages).toBeNull(); + expect(moderationLog?.content).toBeNull(); + expect(moderationLog?.reasoningContent).toBeNull(); + }); + test("/v1/moderations e2e success", async () => { await db.insert(tables.apiKey).values({ id: "token-id", @@ -2889,6 +3068,48 @@ describe("api", () => { expect(logs).toHaveLength(1); }); + test("/v1/images/edits client-error log omits payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id-image-edit-retention-none", + token: "real-token-image-edit-retention-none", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + const requestId = "image-edit-retention-none-request"; + const oversizedImageDataUrl = `data:image/png;base64,${"A".repeat(28 * 1024 * 1024)}`; + + const res = await app.request("/v1/images/edits", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-image-edit-retention-none", + "x-request-id": requestId, + }, + body: JSON.stringify({ + model: "gemini-3-pro-image-preview", + prompt: "secret retention payload", + images: [{ image_url: oversizedImageDataUrl }], + }), + }); + + expect(res.status).toBe(400); + + const log = await waitForLogByRequestId(requestId); + // The client-error log is still recorded... + expect(log.finishReason).toBe("client_error"); + expect(log.hasError).toBe(true); + // ...but the prompt is never persisted for a non-retaining org. + expect(log.messages).toBeNull(); + expect(log.content).toBeNull(); + }); + test("/v1/images/generations forwards X-No-Fallback to chat completions", async () => { await db.insert(tables.apiKey).values({ id: "token-id-image-no-fallback", diff --git a/apps/gateway/src/chat/chat.ts b/apps/gateway/src/chat/chat.ts index 9f8e614e19..82656e8584 100644 --- a/apps/gateway/src/chat/chat.ts +++ b/apps/gateway/src/chat/chat.ts @@ -1991,6 +1991,13 @@ chat.openapi(completions, async (c) => { }); } + // Organization data retention level. Captured here (right after the org is + // resolved) so every log path — including the early rejections below and the + // insertLog wrapper further down — can decide whether payload fields should + // be persisted. Orgs backing end-user wallets are always regular PAYG orgs, + // so the withWalletCredits substitution below never changes this value. + const retentionLevel = organization.retentionLevel ?? "none"; + // Note: the end-user-wallet credits substitution (withWalletCredits) happens // further below — orgs backing end-user wallets are always regular // PAYG/credits orgs, never dev-plan orgs, so it cannot affect the dev-plan @@ -2151,7 +2158,7 @@ chat.openapi(completions, async (c) => { usedServiceTier: null, dataStorageCost: "0", }, - { syncInsert: syncLogInsert }, + { syncInsert: syncLogInsert, retentionLevel }, ); } catch (error) { logger.error("Failed to log unsupported service tier rejection", { @@ -2627,7 +2634,7 @@ chat.openapi(completions, async (c) => { usedServiceTier: null, dataStorageCost: "0", }, - { syncInsert: syncLogInsert }, + { syncInsert: syncLogInsert, retentionLevel }, ); } catch (error) { logger.error("Failed to log budget-thinking rejection", { @@ -2652,9 +2659,6 @@ chat.openapi(completions, async (c) => { let usedRegion: string | undefined = requestedRegion; let routingMetadata: RoutingMetadata | undefined; - // Extract retention level for data storage cost calculation - const retentionLevel = organization?.retentionLevel ?? "none"; - // Get image size limits from environment variables or use defaults const freeLimitMB = Number(process.env.IMAGE_SIZE_LIMIT_FREE_MB) || 50; const proLimitMB = Number(process.env.IMAGE_SIZE_LIMIT_PRO_MB) || 100; @@ -5100,7 +5104,10 @@ chat.openapi(completions, async (c) => { gatewayContentFilterResponse: logData.gatewayContentFilterResponse ?? gatewayContentFilterResponse, }, - options, + // Default the retention level from the resolved organization so payload + // fields are stripped before publishing to the log queue for + // non-retaining orgs. A caller-supplied value still wins. + { retentionLevel, ...options }, ); if (contentFilterBlocked) { diff --git a/apps/gateway/src/embeddings/embeddings.spec.ts b/apps/gateway/src/embeddings/embeddings.spec.ts index 1719d52e36..cd937dbb2c 100644 --- a/apps/gateway/src/embeddings/embeddings.spec.ts +++ b/apps/gateway/src/embeddings/embeddings.spec.ts @@ -7,7 +7,7 @@ import { createGatewayApiTestHarness } from "@/test-utils/gateway-api-test-harne import { resetFailOnceCounter } from "@/test-utils/mock-openai-server.js"; import { waitForLogs } from "@/test-utils/test-helpers.js"; -import { db, tables } from "@llmgateway/db"; +import { db, eq, tables } from "@llmgateway/db"; describe("embeddings", () => { const harness = createGatewayApiTestHarness(); @@ -131,6 +131,55 @@ describe("embeddings", () => { expect(routing?.[0].apiKeyHash).not.toBe(routing?.[1].apiKeyHash); }); + test("/v1/embeddings does not persist payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id-embeddings-retention-none", + token: "real-token-embeddings-retention-none", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-embeddings-retention-none", + token: "openai-token", + provider: "openai", + organizationId: "org-id", + baseUrl: harness.mockServerUrl, + }); + + const res = await app.request("/v1/embeddings", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-embeddings-retention-none", + }, + body: JSON.stringify({ + model: "text-embedding-3-small", + input: "secret retention payload", + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const embeddingLog = logs.find( + (l) => l.usedModel === "openai/text-embedding-3-small", + ); + expect(embeddingLog).toBeDefined(); + expect(embeddingLog?.hasError).toBe(false); + expect(embeddingLog?.finishReason).toBe("stop"); + // Payload never reaches the database for a non-retaining org. + expect(embeddingLog?.messages).toBeNull(); + expect(embeddingLog?.content).toBeNull(); + expect(embeddingLog?.reasoningContent).toBeNull(); + }); + test("/v1/embeddings returns upstream error when no alternate key is available", async () => { resetFailOnceCounter(); diff --git a/apps/gateway/src/embeddings/embeddings.ts b/apps/gateway/src/embeddings/embeddings.ts index f1385e56c1..8afc3db277 100644 --- a/apps/gateway/src/embeddings/embeddings.ts +++ b/apps/gateway/src/embeddings/embeddings.ts @@ -616,6 +616,7 @@ embeddings.openapi(createEmbeddings, async (c): Promise => { } const retentionLevel = organization.retentionLevel ?? "none"; + const iamValidation = await validateRequestModelAccess({ apiKey, organizationId: project.organizationId, @@ -1134,57 +1135,60 @@ embeddings.openapi(createEmbeddings, async (c): Promise => { ); } - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: 0, - content: null, - reasoningContent: null, - finishReason: isCanceled ? "canceled" : "upstream_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: !isCanceled, - streamed: false, - canceled: isCanceled, - errorDetails: isCanceled - ? null - : { - statusCode: 0, - statusText: fetchError.name, - responseText: fetchError.message, - }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: 0, + content: null, + reasoningContent: null, + finishReason: isCanceled ? "canceled" : "upstream_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: !isCanceled, + streamed: false, + canceled: isCanceled, + errorDetails: isCanceled + ? null + : { + statusCode: 0, + statusText: fetchError.name, + responseText: fetchError.message, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, + }, + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -1280,55 +1284,58 @@ embeddings.openapi(createEmbeddings, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(upstreamJson), - reasoningContent: null, - finishReason, - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: status, - statusText: upstreamResponse.statusText, - responseText: upstreamText, + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(upstreamJson), + reasoningContent: null, + finishReason, + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: status, + statusText: upstreamResponse.statusText, + responseText: upstreamText, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -1541,49 +1548,52 @@ embeddings.openapi(createEmbeddings, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(normalizedResponse), - reasoningContent: null, - finishReason: "stop", - promptTokens: promptTokens !== null ? promptTokens.toString() : null, - completionTokens: null, - totalTokens: totalTokens !== null ? totalTokens.toString() : null, - reasoningTokens: null, - cachedTokens: null, - hasError: false, - streamed: false, - canceled: false, - errorDetails: null, - inputCost, - outputCost: 0, - cachedInputCost: 0, - requestCost, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost, - estimatedCost: estimatedUsage, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - promptTokens, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildEmbeddingRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(normalizedResponse), + reasoningContent: null, + finishReason: "stop", + promptTokens: promptTokens !== null ? promptTokens.toString() : null, + completionTokens: null, + totalTokens: totalTokens !== null ? totalTokens.toString() : null, + reasoningTokens: null, + cachedTokens: null, + hasError: false, + streamed: false, + canceled: false, + errorDetails: null, + inputCost, + outputCost: 0, + cachedInputCost: 0, + requestCost, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost, + estimatedCost: estimatedUsage, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + promptTokens, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + }, + { retentionLevel }, + ); return c.json(normalizedResponse); } diff --git a/apps/gateway/src/images/images.ts b/apps/gateway/src/images/images.ts index 6dfd549de9..1fa67e374e 100644 --- a/apps/gateway/src/images/images.ts +++ b/apps/gateway/src/images/images.ts @@ -5,7 +5,11 @@ import { app } from "@/app.js"; import { createLogEntry } from "@/chat/tools/create-log-entry.js"; import { extractCustomHeaders } from "@/chat/tools/extract-custom-headers.js"; import { internalApiOriginHeaders } from "@/lib/api-origin.js"; -import { findApiKeyByToken, findProjectById } from "@/lib/cached-queries.js"; +import { + findApiKeyByToken, + findOrganizationById, + findProjectById, +} from "@/lib/cached-queries.js"; import { parseApiToken } from "@/lib/extract-api-token.js"; import { calculateDataStorageCost, insertLog } from "@/lib/logs.js"; import { validateModelOutput } from "@/lib/validate-model-output.js"; @@ -82,6 +86,7 @@ interface ImageClientErrorLogContext { apiKey: NonNullable>>; project: NonNullable>>; requestId: string; + retentionLevel: "retain" | "none"; } const imageGenerationsResponseSchema = z.object({ @@ -445,6 +450,8 @@ async function resolveImageClientErrorLogContext( return null; } + const organization = await findOrganizationById(project.organizationId); + const requestId = c.req.header("x-request-id")?.trim() || shortid(40); c.header("x-request-id", requestId); @@ -452,6 +459,7 @@ async function resolveImageClientErrorLogContext( apiKey, project, requestId, + retentionLevel: organization?.retentionLevel ?? "none", }; } @@ -485,73 +493,76 @@ async function logImageClientError( } : undefined; - await insertLog({ - ...createLogEntry({ - requestId: logContext.requestId, - project: logContext.project, - apiKey: logContext.apiKey, - usedModel, - usedProvider: "llmgateway", - requestedModel, - messages: [ - { - role: "user", - content: request.prompt ?? "", - }, - ], - source: c.req.header("x-source") ?? undefined, - apiOrigin: "images", - customHeaders: extractCustomHeaders(c), - debugMode: false, - userAgent: c.req.header("user-agent"), - imageConfig, - }), - duration: Date.now() - startedAt, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: responseText.length, - content: null, - reasoningContent: null, - finishReason: "client_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - cacheWriteTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: status, - statusText: getStatusText(status), - responseText, + await insertLog( + { + ...createLogEntry({ + requestId: logContext.requestId, + project: logContext.project, + apiKey: logContext.apiKey, + usedModel, + usedProvider: "llmgateway", + requestedModel, + messages: [ + { + role: "user", + content: request.prompt ?? "", + }, + ], + source: c.req.header("x-source") ?? undefined, + apiOrigin: "images", + customHeaders: extractCustomHeaders(c), + debugMode: false, + userAgent: c.req.header("user-agent"), + imageConfig, + }), + duration: Date.now() - startedAt, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: responseText.length, + content: null, + reasoningContent: null, + finishReason: "client_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + cacheWriteTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: status, + statusText: getStatusText(status), + responseText, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + cacheWriteInputCost: 0, + requestCost: 0, + webSearchCost: 0, + contentFilterCost: null, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + audioInputTokens: null, + audioInputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + requestedServiceTier: null, + usedServiceTier: null, + dataStorageCost: calculateDataStorageCost(null, null, null, null), + cached: false, + tools: null, + toolResults: null, + toolChoice: null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - cacheWriteInputCost: 0, - requestCost: 0, - webSearchCost: 0, - contentFilterCost: null, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - audioInputTokens: null, - audioInputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - requestedServiceTier: null, - usedServiceTier: null, - dataStorageCost: calculateDataStorageCost(null, null, null, null), - cached: false, - tools: null, - toolResults: null, - toolChoice: null, - }); + { retentionLevel: logContext.retentionLevel }, + ); } catch (error) { logger.warn("Images API - failed to log client error", { err: toError(error), diff --git a/apps/gateway/src/lib/logs-retention.spec.ts b/apps/gateway/src/lib/logs-retention.spec.ts new file mode 100644 index 0000000000..bf86fcea6b --- /dev/null +++ b/apps/gateway/src/lib/logs-retention.spec.ts @@ -0,0 +1,135 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { insertLog } from "./logs.js"; + +import type { LogInsertData } from "@llmgateway/db"; + +const publishToQueue = vi.fn(); +const dbInsertValues = vi.fn(); + +vi.mock(import("@llmgateway/cache"), async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + publishToQueue: (...args: unknown[]) => publishToQueue(...args), + }; +}); + +vi.mock(import("@llmgateway/db"), async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + db: { + insert: () => ({ + values: (...args: unknown[]) => dbInsertValues(...args), + }), + } as unknown as typeof actual.db, + }; +}); + +vi.mock("@llmgateway/instrumentation", () => ({ + recordChatCompletionMetrics: vi.fn(), +})); + +function baseLogData(overrides: Partial): LogInsertData { + return { + requestId: "req-1", + organizationId: "org-1", + projectId: "project-1", + apiKeyId: "api-key-1", + duration: 100, + requestedModel: "some-model", + usedModel: "some-model", + usedProvider: "openai", + responseSize: 0, + mode: "credits", + usedMode: "credits", + finishReason: "stop", + hasError: false, + messages: [{ role: "user", content: "secret prompt" }], + content: "secret completion", + reasoningContent: "secret reasoning", + tools: [{ type: "function", function: { name: "f" } }], + toolChoice: "auto", + toolResults: [{ id: "1" }], + responsesApiData: { foo: "bar" }, + ...overrides, + } as LogInsertData; +} + +describe("insertLog retention stripping", () => { + beforeEach(() => { + publishToQueue.mockClear(); + dbInsertValues.mockClear(); + }); + + it("strips payload fields before publishing when retention is none", async () => { + await insertLog(baseLogData({}), { retentionLevel: "none" }); + + expect(publishToQueue).toHaveBeenCalledTimes(1); + const published = publishToQueue.mock.calls[0][1] as LogInsertData; + expect(published.messages).toBeNull(); + expect(published.content).toBeNull(); + expect(published.reasoningContent).toBeNull(); + expect(published.tools).toBeNull(); + expect(published.toolChoice).toBeNull(); + expect(published.toolResults).toBeNull(); + expect(published.responsesApiData).toBeNull(); + // Metadata is preserved. + expect(published.requestId).toBe("req-1"); + expect(published.organizationId).toBe("org-1"); + expect(JSON.stringify(published)).not.toContain("secret"); + }); + + it("keeps payload fields when retention is retain", async () => { + await insertLog(baseLogData({}), { retentionLevel: "retain" }); + + const published = publishToQueue.mock.calls[0][1] as LogInsertData; + expect(published.content).toBe("secret completion"); + expect(published.messages).toEqual([ + { role: "user", content: "secret prompt" }, + ]); + }); + + it("strips payload fields when retention is unspecified (fail closed)", async () => { + await insertLog(baseLogData({})); + + const published = publishToQueue.mock.calls[0][1] as LogInsertData; + expect(published.messages).toBeNull(); + expect(published.content).toBeNull(); + expect(published.reasoningContent).toBeNull(); + }); + + it("strips payload fields when retention is explicitly null (fail closed)", async () => { + await insertLog(baseLogData({}), { retentionLevel: null }); + + const published = publishToQueue.mock.calls[0][1] as LogInsertData; + expect(published.messages).toBeNull(); + expect(published.content).toBeNull(); + }); + + it("strips before a synchronous DB insert when retention is none", async () => { + await insertLog(baseLogData({}), { + syncInsert: true, + retentionLevel: "none", + }); + + expect(dbInsertValues).toHaveBeenCalledTimes(1); + expect(publishToQueue).not.toHaveBeenCalled(); + const inserted = dbInsertValues.mock.calls[0][0] as LogInsertData; + expect(inserted.messages).toBeNull(); + expect(inserted.content).toBeNull(); + expect(inserted.tools).toBeNull(); + expect(inserted.responsesApiData).toBeNull(); + }); + + it("keeps payload on a synchronous DB insert when retention is retain", async () => { + await insertLog(baseLogData({}), { + syncInsert: true, + retentionLevel: "retain", + }); + + const inserted = dbInsertValues.mock.calls[0][0] as LogInsertData; + expect(inserted.content).toBe("secret completion"); + }); +}); diff --git a/apps/gateway/src/lib/logs.ts b/apps/gateway/src/lib/logs.ts index 3189b3365b..35c072680b 100644 --- a/apps/gateway/src/lib/logs.ts +++ b/apps/gateway/src/lib/logs.ts @@ -2,6 +2,7 @@ import { publishToQueue, LOG_QUEUE } from "@llmgateway/cache"; import { db, log, + stripRetentionSensitiveLogFields, UnifiedFinishReason, type LogInsertData, } from "@llmgateway/db"; @@ -287,8 +288,18 @@ export type LogData = InferInsertModel; export async function insertLog( logData: LogInsertData, - options?: { syncInsert?: boolean }, + options?: { syncInsert?: boolean; retentionLevel?: "retain" | "none" | null }, ): Promise { + // Fail closed on retention: unless the organization is explicitly known to + // retain data, strip the request/response payload fields here — before the + // row is ever published to the log queue — so large prompts, completions, and + // tool payloads never travel through Redis. Any omitted, null, or unresolved + // retention level is treated as non-retaining, since the worker no longer + // performs a fallback strip and this is the last chance to withhold payloads. + if (options?.retentionLevel !== "retain") { + logData = stripRetentionSensitiveLogFields(logData); + } + // Stealth providers: the raw upstream error may reveal the underlying // platform, so it survives only in internalErrorDetails — a column excluded // from public API routes and the UI — while the public errorDetails keeps diff --git a/apps/gateway/src/moderations/moderations.ts b/apps/gateway/src/moderations/moderations.ts index 7bd4bb1a38..7d6ec5fe48 100644 --- a/apps/gateway/src/moderations/moderations.ts +++ b/apps/gateway/src/moderations/moderations.ts @@ -621,56 +621,60 @@ moderations.openapi(createModeration, async (c): Promise => { const isTimeout = isTimeoutError(error); const willRetry = !isCanceled && rotateToNextEnvKey(); - await insertLog({ - ...baseLogEntry, - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: 0, - content: null, - reasoningContent: null, - finishReason: isCanceled ? "canceled" : "upstream_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: !isCanceled, - streamed: false, - canceled: isCanceled, - errorDetails: isCanceled - ? null - : { - statusCode: 0, - statusText: error instanceof Error ? error.name : "FetchError", - responseText: - error instanceof Error ? error.message : String(error), - }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + await insertLog( + { + ...baseLogEntry, + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: 0, + content: null, + reasoningContent: null, + finishReason: isCanceled ? "canceled" : "upstream_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: !isCanceled, + streamed: false, + canceled: isCanceled, + errorDetails: isCanceled + ? null + : { + statusCode: 0, + statusText: + error instanceof Error ? error.name : "FetchError", + responseText: + error instanceof Error ? error.message : String(error), + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, + }, + { retentionLevel }, + ); if (willRetry) { continue; @@ -748,53 +752,56 @@ moderations.openapi(createModeration, async (c): Promise => { upstreamText, ) && rotateToNextEnvKey(); - await insertLog({ - ...baseLogEntry, - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(upstreamJson), - reasoningContent: null, - finishReason, - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: upstreamResponse.status, - statusText: upstreamResponse.statusText, - responseText: upstreamText, + await insertLog( + { + ...baseLogEntry, + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(upstreamJson), + reasoningContent: null, + finishReason, + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: upstreamResponse.status, + statusText: upstreamResponse.statusText, + responseText: upstreamText, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + { retentionLevel }, + ); if (willRetry) { continue; @@ -832,48 +839,51 @@ moderations.openapi(createModeration, async (c): Promise => { reportTrackedKeySuccess(providerKey.id); } - await insertLog({ - ...baseLogEntry, - id: finalLogId, - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(upstreamJson), - reasoningContent: null, - finishReason: "stop", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: false, - streamed: false, - canceled: false, - errorDetails: null, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(upstreamJson), + reasoningContent: null, + finishReason: "stop", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: false, + streamed: false, + canceled: false, + errorDetails: null, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + }, + { retentionLevel }, + ); return c.json(upstreamJson as any); } diff --git a/apps/gateway/src/ocr/ocr.spec.ts b/apps/gateway/src/ocr/ocr.spec.ts index 4065adcae7..94aa7431c8 100644 --- a/apps/gateway/src/ocr/ocr.spec.ts +++ b/apps/gateway/src/ocr/ocr.spec.ts @@ -4,7 +4,7 @@ import { app } from "@/app.js"; import { createGatewayApiTestHarness } from "@/test-utils/gateway-api-test-harness.js"; import { waitForLogs } from "@/test-utils/test-helpers.js"; -import { db, tables } from "@llmgateway/db"; +import { db, eq, tables } from "@llmgateway/db"; const IMAGE_DOCUMENT = { type: "image_url" as const, @@ -14,6 +14,58 @@ const IMAGE_DOCUMENT = { describe("ocr", () => { const harness = createGatewayApiTestHarness(); + test("/v1/ocr does not persist payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id-ocr-retention-none", + token: "real-token-ocr-retention-none", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-ocr-retention-none", + token: "mistral-token", + provider: "mistral", + organizationId: "org-id", + baseUrl: harness.mockServerUrl, + }); + + const res = await app.request("/v1/ocr", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-ocr-retention-none", + }, + body: JSON.stringify({ + model: "mistral-ocr-latest", + document: { + type: "image_url", + image_url: "https://example.com/PAGES_3.png", + }, + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const ocrLog = logs.find( + (l) => l.usedModel === "mistral/mistral-ocr-latest", + ); + expect(ocrLog).toBeDefined(); + expect(ocrLog?.hasError).toBe(false); + expect(ocrLog?.finishReason).toBe("stop"); + // Payload never reaches the database for a non-retaining org. + expect(ocrLog?.messages).toBeNull(); + expect(ocrLog?.content).toBeNull(); + expect(ocrLog?.reasoningContent).toBeNull(); + }); + test("/v1/ocr returns the OCR result and bills per page", async () => { await db.insert(tables.apiKey).values({ id: "token-id-ocr", diff --git a/apps/gateway/src/ocr/ocr.ts b/apps/gateway/src/ocr/ocr.ts index 6493b14135..736162032d 100644 --- a/apps/gateway/src/ocr/ocr.ts +++ b/apps/gateway/src/ocr/ocr.ts @@ -508,6 +508,7 @@ ocr.openapi(createOcr, async (c): Promise => { } const retentionLevel = organization.retentionLevel ?? "none"; + const iamValidation = await validateRequestModelAccess({ apiKey, organizationId: project.organizationId, @@ -819,57 +820,60 @@ ocr.openapi(createOcr, async (c): Promise => { ); } - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: 0, - content: null, - reasoningContent: null, - finishReason: isCanceled ? "canceled" : "upstream_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: !isCanceled, - streamed: false, - canceled: isCanceled, - errorDetails: isCanceled - ? null - : { - statusCode: 0, - statusText: fetchError.name, - responseText: fetchError.message, - }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: 0, + content: null, + reasoningContent: null, + finishReason: isCanceled ? "canceled" : "upstream_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: !isCanceled, + streamed: false, + canceled: isCanceled, + errorDetails: isCanceled + ? null + : { + statusCode: 0, + statusText: fetchError.name, + responseText: fetchError.message, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, + }, + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -962,55 +966,58 @@ ocr.openapi(createOcr, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(upstreamJson), - reasoningContent: null, - finishReason, - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: status, - statusText: upstreamResponse.statusText, - responseText: upstreamText, + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(upstreamJson), + reasoningContent: null, + finishReason, + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: status, + statusText: upstreamResponse.statusText, + responseText: upstreamText, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -1087,49 +1094,52 @@ ocr.openapi(createOcr, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: getResponseContent(upstreamJson), - reasoningContent: null, - finishReason: "stop", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: false, - streamed: false, - canceled: false, - errorDetails: null, - inputCost: pageCost, - outputCost: 0, - cachedInputCost: 0, - requestCost, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost, - estimatedCost: pagesProcessed === null, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildOcrRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: getResponseContent(upstreamJson), + reasoningContent: null, + finishReason: "stop", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: false, + streamed: false, + canceled: false, + errorDetails: null, + inputCost: pageCost, + outputCost: 0, + cachedInputCost: 0, + requestCost, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost, + estimatedCost: pagesProcessed === null, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + }, + { retentionLevel }, + ); return c.json(upstreamJson as z.infer); } diff --git a/apps/gateway/src/speech/speech.spec.ts b/apps/gateway/src/speech/speech.spec.ts index adbaa6945f..206e0fa8f7 100644 --- a/apps/gateway/src/speech/speech.spec.ts +++ b/apps/gateway/src/speech/speech.spec.ts @@ -85,6 +85,45 @@ describe("speech", () => { expect(Number(log?.outputCost)).toBeCloseTo(42 * 10e-6, 10); }); + test("/v1/audio/speech does not persist payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await seedKeys( + "real-token-speech-retention-none", + "token-id-speech-retention-none", + ); + + const res = await app.request("/v1/audio/speech", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token-speech-retention-none", + }, + body: JSON.stringify({ + model: "gemini-2.5-flash-preview-tts", + input: "secret retention payload", + voice: "Kore", + }), + }); + + expect(res.status).toBe(200); + + const logs = await waitForLogs(1); + const log = logs.find( + (l) => l.usedModel === "google-ai-studio/gemini-2.5-flash-preview-tts", + ); + expect(log).toBeDefined(); + expect(log?.hasError).toBe(false); + expect(log?.finishReason).toBe("stop"); + // Payload never reaches the database for a non-retaining org. + expect(log?.messages).toBeNull(); + expect(log?.content).toBeNull(); + expect(log?.reasoningContent).toBeNull(); + }); + test("/v1/audio/speech returns raw PCM when requested", async () => { await seedKeys("real-token-speech-pcm", "token-id-speech-pcm"); diff --git a/apps/gateway/src/speech/speech.ts b/apps/gateway/src/speech/speech.ts index 226f81d4d7..8ffe26b926 100644 --- a/apps/gateway/src/speech/speech.ts +++ b/apps/gateway/src/speech/speech.ts @@ -640,6 +640,7 @@ speech.openapi(createSpeech, async (c): Promise => { } const retentionLevel = organization.retentionLevel ?? "none"; + const iamValidation = await validateRequestModelAccess({ apiKey, organizationId: project.organizationId, @@ -1065,57 +1066,60 @@ speech.openapi(createSpeech, async (c): Promise => { ); } - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: 0, - content: null, - reasoningContent: null, - finishReason: isCanceled ? "canceled" : "upstream_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: !isCanceled, - streamed: false, - canceled: isCanceled, - errorDetails: isCanceled - ? null - : { - statusCode: 0, - statusText: fetchError.name, - responseText: fetchError.message, - }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: 0, + content: null, + reasoningContent: null, + finishReason: isCanceled ? "canceled" : "upstream_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: !isCanceled, + streamed: false, + canceled: isCanceled, + errorDetails: isCanceled + ? null + : { + statusCode: 0, + statusText: fetchError.name, + responseText: fetchError.message, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, + }, + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -1207,55 +1211,58 @@ speech.openapi(createSpeech, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: willRetry ? attemptLogId : finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: null, - reasoningContent: null, - finishReason, - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: status, - statusText: upstreamResponse.statusText, - responseText: upstreamText, + await insertLog( + { + ...baseLogEntry, + id: willRetry ? attemptLogId : finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: null, + reasoningContent: null, + finishReason, + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: status, + statusText: upstreamResponse.statusText, + responseText: upstreamText, + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + retried: willRetry, + retriedByLogId: willRetry ? finalLogId : null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - retried: willRetry, - retriedByLogId: willRetry ? finalLogId : null, - }); + { retentionLevel }, + ); if (willRetry && nextAttempt) { attempt = nextAttempt; @@ -1361,53 +1368,56 @@ speech.openapi(createSpeech, async (c): Promise => { { apiKeyHash: usedApiKeyHash, logId: finalLogId }, ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: sseText.length, - content: null, - reasoningContent: null, - finishReason: "upstream_error", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: upstreamResponse.status, - statusText: "no_audio", - responseText: (sseErrorMessage ?? sseText).slice(0, 2000), + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: sseText.length, + content: null, + reasoningContent: null, + finishReason: "upstream_error", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: upstreamResponse.status, + statusText: "no_audio", + responseText: (sseErrorMessage ?? sseText).slice(0, 2000), + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + { retentionLevel }, + ); return c.json( { error: { @@ -1473,50 +1483,54 @@ speech.openapi(createSpeech, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: out.length, - content: `[audio: ${out.length} bytes, ${contentType}]`, - reasoningContent: null, - finishReason: "stop", - promptTokens: promptTokens !== null ? promptTokens.toString() : null, - completionTokens: - completionTokens !== null ? completionTokens.toString() : null, - totalTokens: totalTokens !== null ? totalTokens.toString() : null, - reasoningTokens: null, - cachedTokens: null, - hasError: false, - streamed: false, - canceled: false, - errorDetails: null, - inputCost, - outputCost, - cachedInputCost: 0, - requestCost, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost, - estimatedCost, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - promptTokens, - null, - completionTokens, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: out.length, + content: `[audio: ${out.length} bytes, ${contentType}]`, + reasoningContent: null, + finishReason: "stop", + promptTokens: + promptTokens !== null ? promptTokens.toString() : null, + completionTokens: + completionTokens !== null ? completionTokens.toString() : null, + totalTokens: totalTokens !== null ? totalTokens.toString() : null, + reasoningTokens: null, + cachedTokens: null, + hasError: false, + streamed: false, + canceled: false, + errorDetails: null, + inputCost, + outputCost, + cachedInputCost: 0, + requestCost, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost, + estimatedCost, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + promptTokens, + null, + completionTokens, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + }, + { retentionLevel }, + ); return c.body(toArrayBuffer(out), 200, { "Content-Type": contentType, @@ -1755,53 +1769,56 @@ speech.openapi(createSpeech, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize, - content: null, - reasoningContent: null, - finishReason: "content_filter", - promptTokens: null, - completionTokens: null, - totalTokens: null, - reasoningTokens: null, - cachedTokens: null, - hasError: true, - streamed: false, - canceled: false, - errorDetails: { - statusCode: upstreamResponse.status, - statusText: "no_audio", - responseText: upstreamText.slice(0, 2000), + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize, + content: null, + reasoningContent: null, + finishReason: "content_filter", + promptTokens: null, + completionTokens: null, + totalTokens: null, + reasoningTokens: null, + cachedTokens: null, + hasError: true, + streamed: false, + canceled: false, + errorDetails: { + statusCode: upstreamResponse.status, + statusText: "no_audio", + responseText: upstreamText.slice(0, 2000), + }, + inputCost: 0, + outputCost: 0, + cachedInputCost: 0, + requestCost: 0, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost: 0, + estimatedCost: false, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + null, + null, + null, + null, + retentionLevel, + ), + cached: false, + toolResults: null, }, - inputCost: 0, - outputCost: 0, - cachedInputCost: 0, - requestCost: 0, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost: 0, - estimatedCost: false, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - null, - null, - null, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + { retentionLevel }, + ); return c.json( { @@ -1860,50 +1877,53 @@ speech.openapi(createSpeech, async (c): Promise => { ), ); - await insertLog({ - ...baseLogEntry, - id: finalLogId, - routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), - duration, - timeToFirstToken: null, - timeToFirstReasoningToken: null, - responseSize: out.length, - content: `[audio: ${out.length} bytes, ${audioMimeType ?? "audio/wav"}]`, - reasoningContent: null, - finishReason: "stop", - promptTokens: promptTokens !== null ? promptTokens.toString() : null, - completionTokens: - audioOutputTokens !== null ? audioOutputTokens.toString() : null, - totalTokens: totalTokens !== null ? totalTokens.toString() : null, - reasoningTokens: null, - cachedTokens: null, - hasError: false, - streamed: false, - canceled: false, - errorDetails: null, - inputCost, - outputCost, - cachedInputCost: 0, - requestCost, - webSearchCost: 0, - imageInputTokens: null, - imageOutputTokens: null, - imageInputCost: null, - imageOutputCost: null, - cost, - estimatedCost: promptTokens === null || audioOutputTokens === null, - discount: null, - pricingTier: null, - dataStorageCost: calculateDataStorageCost( - promptTokens, - null, - audioOutputTokens, - null, - retentionLevel, - ), - cached: false, - toolResults: null, - }); + await insertLog( + { + ...baseLogEntry, + id: finalLogId, + routingMetadata: buildSpeechRoutingMetadata(usedApiKeyHash), + duration, + timeToFirstToken: null, + timeToFirstReasoningToken: null, + responseSize: out.length, + content: `[audio: ${out.length} bytes, ${audioMimeType ?? "audio/wav"}]`, + reasoningContent: null, + finishReason: "stop", + promptTokens: promptTokens !== null ? promptTokens.toString() : null, + completionTokens: + audioOutputTokens !== null ? audioOutputTokens.toString() : null, + totalTokens: totalTokens !== null ? totalTokens.toString() : null, + reasoningTokens: null, + cachedTokens: null, + hasError: false, + streamed: false, + canceled: false, + errorDetails: null, + inputCost, + outputCost, + cachedInputCost: 0, + requestCost, + webSearchCost: 0, + imageInputTokens: null, + imageOutputTokens: null, + imageInputCost: null, + imageOutputCost: null, + cost, + estimatedCost: promptTokens === null || audioOutputTokens === null, + discount: null, + pricingTier: null, + dataStorageCost: calculateDataStorageCost( + promptTokens, + null, + audioOutputTokens, + null, + retentionLevel, + ), + cached: false, + toolResults: null, + }, + { retentionLevel }, + ); return c.body(toArrayBuffer(out), 200, { "Content-Type": contentType, diff --git a/apps/gateway/src/videos/videos.spec.ts b/apps/gateway/src/videos/videos.spec.ts index b220dd0837..0d54174ab4 100644 --- a/apps/gateway/src/videos/videos.spec.ts +++ b/apps/gateway/src/videos/videos.spec.ts @@ -1225,6 +1225,77 @@ describe("videos", () => { expect(logs[0].cost).toBe(2.8); }); + test("/v1/videos does not persist payload when retention is disabled", async () => { + await db + .update(tables.organization) + .set({ retentionLevel: "none" }) + .where(eq(tables.organization.id, "org-id")); + + await db.insert(tables.apiKey).values({ + id: "token-id", + token: "real-token", + projectId: "project-id", + description: "Test API Key", + createdBy: "user-id", + }); + + await db.insert(tables.providerKey).values({ + id: "provider-key-id", + token: "sk-test-key", + provider: "avalanche", + organizationId: "org-id", + baseUrl: mockServerUrl, + }); + + const createRes = await app.request("/v1/videos", { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer real-token", + }, + body: JSON.stringify({ + model: "avalanche/veo-3.1-fast-generate-preview", + prompt: "A storm above a mountain range", + size: "3840x2160", + seconds: 8, + }), + }); + + expect(createRes.status).toBe(200); + const created = await createRes.json(); + + const videoJob = await db.query.videoJob.findFirst({ + where: { id: { eq: created.id } }, + }); + expect(videoJob).toBeTruthy(); + + setMockVideoStatus(videoJob!.upstreamId, "completed"); + await processPendingVideoJobs(); + + const logs = await db.query.log.findMany({ + where: { usedModel: { eq: "avalanche/veo-3.1-fast-generate-preview" } }, + }); + expect(logs).toHaveLength(1); + expect(logs[0].hasError).toBe(false); + // Cost/metering is still recorded... + expect(logs[0].videoOutputCost).toBe(2.8); + // ...but no request/response payload is persisted for a non-retaining org. + expect(logs[0].messages).toBeNull(); + expect(logs[0].content).toBeNull(); + + // Video content is served from the video job, not the (now-stripped) log + // content column, so playback still works for a non-retaining org. + const contentRes = await app.request(`/v1/videos/${created.id}/content`, { + headers: { + Authorization: "Bearer real-token", + }, + }); + expect(contentRes.status).toBe(200); + expect(await contentRes.text()).toBe( + `mock-video-${videoJob!.upstreamId}-4k`, + ); + }); + test("/v1/videos bills xAI 480p video and image input separately", async () => { await db.insert(tables.apiKey).values({ id: "token-id", diff --git a/apps/worker/src/services/video-jobs.ts b/apps/worker/src/services/video-jobs.ts index 91248e7085..41d54c722e 100644 --- a/apps/worker/src/services/video-jobs.ts +++ b/apps/worker/src/services/video-jobs.ts @@ -11,9 +11,11 @@ import { type InferSelectModel, inArray, isNull, + type LogInsertData, lte, or, shortid, + stripRetentionSensitiveLogFields, tables, UnifiedFinishReason, } from "@llmgateway/db"; @@ -1822,15 +1824,6 @@ async function finalizeVideoJob(job: VideoJobRecord): Promise { const totalCost = Number((videoOutputCost + imageInputCost).toFixed(6)); const responsePayload = await serializeVideoJob(jobToLog, logId); const responseSize = JSON.stringify(responsePayload).length; - const messages = - organization?.retentionLevel === "retain" - ? [ - { - role: "user", - content: jobToLog.prompt, - }, - ] - : null; const isContentFilterFailure = jobToLog.status === "failed" && @@ -1846,7 +1839,7 @@ async function finalizeVideoJob(job: VideoJobRecord): Promise { ? UnifiedFinishReason.CONTENT_FILTER : UnifiedFinishReason.UPSTREAM_ERROR; - await tx.insert(tables.log).values({ + const logValues: LogInsertData = { id: logId, requestId: jobToLog.requestId, apiOrigin: "videos", @@ -1885,7 +1878,12 @@ async function finalizeVideoJob(job: VideoJobRecord): Promise { imageInputCost, videoOutputCost, estimatedCost: false, - messages, + messages: [ + { + role: "user", + content: jobToLog.prompt, + }, + ], mode: jobToLog.mode, usedMode: jobToLog.usedMode, routingMetadata: jobToLog.routingMetadata ?? null, @@ -1901,7 +1899,17 @@ async function finalizeVideoJob(job: VideoJobRecord): Promise { upstreamResponse: jobToLog.upstreamStatusResponse, processedAt: null, dataStorageCost: "0", - }); + }; + + // Strip request/response payload fields for orgs that don't retain data, + // keeping the video log consistent with every other endpoint's policy. + await tx + .insert(tables.log) + .values( + organization?.retentionLevel === "retain" + ? logValues + : stripRetentionSensitiveLogFields(logValues), + ); await tx .update(tables.videoJob) diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 4fc8348c07..0d4e1f9938 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -15,7 +15,6 @@ import { addApiKeyPeriodDuration, and, apiKey, - cdb, closeDatabase, db, enqueueWebhookDeliveries, @@ -122,11 +121,6 @@ const LOG_QUEUE_CONCURRENCY = Math.max( 1, Number(process.env.LOG_QUEUE_CONCURRENCY) || 4, ); -// Cache organization retention levels to avoid a serial Postgres round-trip -// before every log batch insert. retentionLevel changes rarely; the short TTL -// bounds how long a stale value can keep retaining or stripping log payloads. -const ORG_RETENTION_CACHE_TTL_MS = - Number(process.env.ORG_RETENTION_CACHE_TTL_MS) || 60_000; const CREDIT_BATCH_SIZE = Number(process.env.CREDIT_BATCH_SIZE) || 100; const BATCH_PROCESSING_INTERVAL_SECONDS = Number(process.env.CREDIT_BATCH_INTERVAL) || 5; @@ -1686,50 +1680,6 @@ function recordLogInsertSuccess(): void { logInsertCircuit.nextAttemptAt = 0; } -const orgRetentionCache = new Map< - string, - { retentionLevel: "retain" | "none"; expiresAt: number } ->(); - -// Resolve organization retention levels, serving from the in-memory cache when -// fresh and only querying Postgres for the ids that are missing or expired. -async function getOrganizationRetentionLevels( - organizationIds: string[], -): Promise> { - const now = Date.now(); - const result = new Map(); - const missing: string[] = []; - - for (const id of organizationIds) { - const cached = orgRetentionCache.get(id); - if (cached && cached.expiresAt > now) { - result.set(id, cached.retentionLevel); - } else { - missing.push(id); - } - } - - if (missing.length > 0) { - const organizations = await cdb - .select({ - id: organization.id, - retentionLevel: organization.retentionLevel, - }) - .from(organization) - .where(inArray(organization.id, missing)); - - for (const org of organizations) { - result.set(org.id, org.retentionLevel); - orgRetentionCache.set(org.id, { - retentionLevel: org.retentionLevel, - expiresAt: now + ORG_RETENTION_CACHE_TTL_MS, - }); - } - } - - return result; -} - // Returns the number of messages successfully inserted, so the drain loop can // decide whether to sleep (partial batch) or immediately fetch the next batch // (full batch, queue likely still backed up). @@ -1747,49 +1697,21 @@ export async function processLogQueue(): Promise { const MAX_RETRIES = 5; try { + // The gateway decides what to persist: it strips request/response payload + // fields before publishing for orgs that don't retain data, so the worker + // inserts the queued rows as-is with no per-batch org retention lookup. const logData = message.map((i) => JSON.parse(i) as LogInsertData); - const organizationIds = Array.from( - new Set(logData.map((data) => data.organizationId)), - ); - const selectStart = Date.now(); - const retentionByOrg = - organizationIds.length > 0 - ? await getOrganizationRetentionLevels(organizationIds) - : new Map(); - const selectMs = Date.now() - selectStart; - - const processedLogData: ( - | LogInsertData - | Omit - )[] = logData.map((data) => { - if (retentionByOrg.get(data.organizationId) === "none") { - const { - messages: _messages, - content: _content, - reasoningContent: _reasoningContent, - tools: _tools, - toolChoice: _toolChoice, - toolResults: _toolResults, - responsesApiData: _responsesApiData, - ...metadataOnly - } = data; - return metadataOnly; - } - - return data; - }); // Insert logs with retry logic let lastError: Error | undefined; for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { try { - // Type assertion is safe here as both LogInsertData and its subset are compatible with the log insert schema const insertStart = Date.now(); - await db.insert(log).values(processedLogData as LogInsertData[]); + await db.insert(log).values(logData); const insertMs = Date.now() - insertStart; recordLogInsertSuccess(); logger.info( - `Processed log batch: ${message.length} rows (org lookup ${selectMs}ms, insert ${insertMs}ms)`, + `Processed log batch: ${message.length} rows (insert ${insertMs}ms)`, ); return message.length; // Success, exit function } catch (insertError) { diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 1c851c3623..480bfd7cab 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -10,6 +10,7 @@ export * from "./email-recipients.js"; export * from "./rate-limit-helpers.js"; export * from "./schema.js"; export * from "./log-payloads.js"; +export * from "./log-retention.js"; export * from "./types.js"; export * from "./migrate.js"; export * from "./relations.js"; diff --git a/packages/db/src/log-retention.spec.ts b/packages/db/src/log-retention.spec.ts new file mode 100644 index 0000000000..b0370b1b5d --- /dev/null +++ b/packages/db/src/log-retention.spec.ts @@ -0,0 +1,103 @@ +import { describe, expect, it } from "vitest"; + +import { + RETENTION_SENSITIVE_LOG_FIELDS, + stripRetentionSensitiveLogFields, +} from "./log-retention.js"; + +import type { LogInsertData } from "./types.js"; + +function baseLogData(overrides: Partial = {}): LogInsertData { + return { + requestId: "req-1", + organizationId: "org-1", + projectId: "project-1", + apiKeyId: "api-key-1", + duration: 100, + requestedModel: "some-model", + usedModel: "some-model", + usedProvider: "openai", + responseSize: 42, + mode: "credits", + usedMode: "credits", + finishReason: "stop", + hasError: false, + promptTokens: "10", + completionTokens: "20", + messages: [{ role: "user", content: "secret prompt" }], + content: "secret completion", + reasoningContent: "secret reasoning", + tools: [{ type: "function", function: { name: "f" } }], + toolChoice: "auto", + toolResults: [{ id: "1" }], + responsesApiData: { foo: "bar" }, + ...overrides, + } as LogInsertData; +} + +describe("stripRetentionSensitiveLogFields", () => { + it("clears every retention-sensitive payload field", () => { + const stripped = stripRetentionSensitiveLogFields(baseLogData()); + + expect(stripped.messages).toBeNull(); + expect(stripped.content).toBeNull(); + expect(stripped.reasoningContent).toBeNull(); + expect(stripped.tools).toBeNull(); + expect(stripped.toolChoice).toBeNull(); + expect(stripped.toolResults).toBeNull(); + expect(stripped.responsesApiData).toBeNull(); + }); + + it("nulls exactly the documented field list and nothing else", () => { + const input = baseLogData(); + const stripped = stripRetentionSensitiveLogFields(input); + + for (const key of Object.keys(input) as (keyof LogInsertData)[]) { + if ( + (RETENTION_SENSITIVE_LOG_FIELDS as readonly string[]).includes( + key as string, + ) + ) { + expect(stripped[key]).toBeNull(); + } else { + expect(stripped[key]).toEqual(input[key]); + } + } + }); + + it("preserves all metadata / metering fields", () => { + const stripped = stripRetentionSensitiveLogFields(baseLogData()); + + expect(stripped.requestId).toBe("req-1"); + expect(stripped.organizationId).toBe("org-1"); + expect(stripped.projectId).toBe("project-1"); + expect(stripped.usedProvider).toBe("openai"); + expect(stripped.responseSize).toBe(42); + expect(stripped.promptTokens).toBe("10"); + expect(stripped.completionTokens).toBe("20"); + expect(stripped.finishReason).toBe("stop"); + }); + + it("does not mutate the input object", () => { + const input = baseLogData(); + const strippedContent = input.content; + stripRetentionSensitiveLogFields(input); + + expect(input.content).toBe(strippedContent); + expect(input.messages).toEqual([ + { role: "user", content: "secret prompt" }, + ]); + }); + + it("is idempotent on already-stripped data", () => { + const once = stripRetentionSensitiveLogFields(baseLogData()); + const twice = stripRetentionSensitiveLogFields(once); + + expect(twice).toEqual(once); + }); + + it("leaves no secret payload strings in the result", () => { + const stripped = stripRetentionSensitiveLogFields(baseLogData()); + expect(JSON.stringify(stripped)).not.toContain("secret"); + }); +}); diff --git a/packages/db/src/log-retention.ts b/packages/db/src/log-retention.ts new file mode 100644 index 0000000000..2cc8cf09e6 --- /dev/null +++ b/packages/db/src/log-retention.ts @@ -0,0 +1,38 @@ +import type { LogInsertData } from "./types.js"; + +/** + * Log columns that hold request/response payload data (the actual prompt, + * completion, reasoning, tool definitions/results and Responses API payload). + * Organizations with `retentionLevel` "none" do not retain any of this, so it + * is cleared before the row is persisted. + */ +export const RETENTION_SENSITIVE_LOG_FIELDS = [ + "messages", + "content", + "reasoningContent", + "tools", + "toolChoice", + "toolResults", + "responsesApiData", +] as const; + +/** + * Return a copy of the log data with every retention-sensitive payload field + * cleared. Used by the gateway so payloads for non-retaining orgs never travel + * through Redis (or reach the database) in the first place — the gateway is the + * sole decider of what gets persisted, so the worker inserts queued rows as-is. + */ +export function stripRetentionSensitiveLogFields( + logData: T, +): T { + return { + ...logData, + messages: null, + content: null, + reasoningContent: null, + tools: null, + toolChoice: null, + toolResults: null, + responsesApiData: null, + }; +}