diff --git a/platform/e2e/pages/models-page.ts b/platform/e2e/pages/models-page.ts index e58587d6..1e31b022 100644 --- a/platform/e2e/pages/models-page.ts +++ b/platform/e2e/pages/models-page.ts @@ -52,6 +52,7 @@ export function modelsPage(page: Page) { }) await page.getByRole("button", { name: "Save provider", exact: true }).click() await expect(page.getByRole("heading", { level: 1, name: input.name })).toBeVisible() + await expectToast(page, "Provider saved") await expect(page.locator("body")).not.toHaveAttribute("data-key-error-shown") }, diff --git a/platform/e2e/specs/seeded/model-price-schedule.spec.ts b/platform/e2e/specs/seeded/model-price-schedule.spec.ts new file mode 100644 index 00000000..3a719a10 --- /dev/null +++ b/platform/e2e/specs/seeded/model-price-schedule.spec.ts @@ -0,0 +1,120 @@ +import { + createSeedDatabase, + loadSeedEnvironment, + resolveSeedDatabaseUrl, +} from "@scripts/seed/database.ts" +import { SEED_ORGANIZATIONS, SEED_PASSWORD, SEED_USERS } from "@scripts/seed/fixtures.ts" +import { captureMilestone } from "@e2e/capture.ts" +import { expect, test } from "@e2e/fixtures.ts" +import { waitForHydration } from "@e2e/hydration.ts" + +test("manual prices can be scheduled, edited and cancelled while viewers cannot manage prices", async ({ + page, + models, + auth, +}) => { + test.setTimeout(90_000) + const org = SEED_ORGANIZATIONS[0] + const model = "gpt-6.1-sol" + const name = `Pricing schedule ${Date.now()}` + await page.goto(`/${org.slug}/models`) + await models.create({ name, modelId: model, apiKey: "synthetic-scheduling-key" }) + const providerUrl = page.url() + await page + .getByRole("switch", { name: `Override catalog defaults for ${model}`, exact: true }) + .check() + await page.getByLabel(`Input price for ${model}`, { exact: true }).fill("2") + await page.getByLabel(`Output price for ${model}`, { exact: true }).fill("8") + await models.save() + await page.reload() + await waitForHydration(page.locator("#model-provider-name")) + const schedule = page.getByRole("group", { name: `Price schedule for ${model}`, exact: true }) + await schedule.getByRole("button", { name: "Schedule price change" }).click() + const future = new Date(Date.now() + 86_400_000) + const local = new Date(future.getTime() - future.getTimezoneOffset() * 60_000) + .toISOString() + .slice(0, 16) + await schedule.getByLabel("Effective from", { exact: true }).fill(local) + await schedule.getByLabel(/Input price/).fill("3") + await schedule.getByLabel(/Output price/).fill("12") + await models.save() + await page.reload() + await waitForHydration(page.locator("#model-provider-name")) + await expect(page.getByLabel(`Input price for ${model}`, { exact: true })).toHaveValue("2") + await schedule.locator("summary").filter({ hasText: "Scheduled" }).click() + await expect(schedule.getByLabel("Effective from", { exact: true })).toHaveValue(local) + await expect(schedule.getByLabel(/Input price/)).toHaveValue("3") + await schedule.scrollIntoViewIfNeeded() + await captureMilestone(page, "01-saved-price-schedule") + await schedule.getByLabel(/Input price/).fill("4") + await models.save() + await page.reload() + await waitForHydration(page.locator("#model-provider-name")) + await schedule.locator("summary").filter({ hasText: "Scheduled" }).click() + await expect(schedule.getByLabel(/Input price/)).toHaveValue("4") + await page.setViewportSize({ width: 390, height: 844 }) + await schedule.scrollIntoViewIfNeeded() + expect(await page.evaluate(() => document.documentElement.scrollWidth <= window.innerWidth)).toBe( + true, + ) + await captureMilestone(page, "02-edited-price-schedule-mobile") + await page.setViewportSize({ width: 1440, height: 900 }) + await page.goto("/auth/sign-out") + await page.waitForURL(/\/auth\/sign-in/) + await auth.signIn(SEED_USERS[2].email, SEED_PASSWORD) + await page.waitForURL((url) => !url.pathname.startsWith("/auth/")) + await page.goto(providerUrl) + await expect(page).toHaveURL(new RegExp(`/${org.slug}/?$`)) + await expect(schedule).toHaveCount(0) + await expect(page.getByRole("button", { name: "Save provider", exact: true })).toHaveCount(0) + await captureMilestone(page, "03-viewer-cannot-manage-prices") + await page.goto("/auth/sign-out") + await page.waitForURL(/\/auth\/sign-in/) + await auth.signIn(SEED_USERS[0].email, SEED_PASSWORD) + await page.waitForURL((url) => !url.pathname.startsWith("/auth/")) + await page.goto(providerUrl) + await waitForHydration(page.locator("#model-provider-name")) + await schedule.locator("summary").filter({ hasText: "Scheduled" }).click() + await schedule.getByRole("button", { name: "Cancel scheduled price" }).click() + await models.save() + await page.reload() + await waitForHydration(page.locator("#model-provider-name")) + await expect(schedule.locator("details")).toHaveCount(0) + await expect(page.getByLabel(`Input price for ${model}`, { exact: true })).toHaveValue("2") + await schedule.scrollIntoViewIfNeeded() + await captureMilestone(page, "04-price-schedule-cancelled") + await schedule.getByRole("button", { name: "Schedule price change" }).click() + await schedule.getByLabel("Effective from", { exact: true }).fill(local) + await schedule.getByLabel(/Input price/).fill("4") + await models.save() + loadSeedEnvironment() + const { pool } = createSeedDatabase(resolveSeedDatabaseUrl().url) + try { + // Activate this test's saved schedule without changing the provider's lock version. + const activated = await pool.query( + `update provider_model + set pricing_history = jsonb_set(pricing_history, '{-1,effectiveFrom}', to_jsonb($4::text)), + updated_at = now() + where organization_id = $1 and model_provider_id = $2 and model_id = $3`, + [ + org.id, + new URL(providerUrl).pathname.split("/").filter(Boolean).at(-1), + model, + new Date(Date.now() - 1).toISOString(), + ], + ) + expect(activated.rowCount).toBe(1) + } finally { + await pool.end() + } + await page.getByLabel(`Input price for ${model}`, { exact: true }).fill("5") + await page.getByRole("button", { name: "Save provider", exact: true }).click() + await expect(page.getByRole("button", { name: "Save provider", exact: true })).toBeDisabled() + await page.getByRole("button", { name: "Reload latest", exact: true }).click() + await expect(page.getByLabel(`Input price for ${model}`, { exact: true })).toHaveValue("4") + await waitForHydration(page.locator("#model-provider-name")) + await expect(schedule.locator("details")).toHaveCount(0) + await models.save() + await captureMilestone(page, "05-activated-price-reloaded") + await models.requestDelete() +}) diff --git a/platform/scripts/check-binary.ts b/platform/scripts/check-binary.ts index e4f671c5..fdfc79fe 100644 --- a/platform/scripts/check-binary.ts +++ b/platform/scripts/check-binary.ts @@ -17,6 +17,8 @@ import packageJson from "../package.json" with { type: "json" } const BINARY_CHECK_START_TIMEOUT_MS = 20_000 const BINARY_CHECK_FETCH_TIMEOUT_MS = 2_000 +// Shutdown has separate five-second HTTP drain and resource cleanup budgets. +const BINARY_CHECK_SHUTDOWN_TIMEOUT_MS = 11_000 const runBinaryCheckCommand = promisify(execFile) async function reserveBinaryCheckPort() { @@ -151,7 +153,10 @@ async function stopBinaryCheckProcess(binaryProcess: ChildProcess, status: Promi const processId = binaryProcess.pid if (processId === undefined) throw new Error("Binary process did not start") process.kill(processId, "SIGTERM") - const stopped = await Promise.race([status.then(() => true), delay(5_000, false)]) + const stopped = await Promise.race([ + status.then(() => true), + delay(BINARY_CHECK_SHUTDOWN_TIMEOUT_MS, false), + ]) if (stopped) { if (binaryProcess.exitCode !== 0) { throw new Error("Binary did not exit cleanly after SIGTERM") @@ -160,7 +165,7 @@ async function stopBinaryCheckProcess(binaryProcess: ChildProcess, status: Promi } process.kill(processId, "SIGKILL") await status - throw new Error("Binary did not exit within 5 seconds of SIGTERM") + throw new Error("Binary did not exit within 11 seconds of SIGTERM") } async function runBinaryCheck() { diff --git a/platform/scripts/seed/models.ts b/platform/scripts/seed/models.ts index a17a9ad5..f0d397ca 100644 --- a/platform/scripts/seed/models.ts +++ b/platform/scripts/seed/models.ts @@ -60,6 +60,13 @@ export async function seedModelProviders( modelProviderId, modelId: SEED_MODEL_PROVIDER.modelId, name: SEED_MODEL_PROVIDER.modelName, + pricingHistory: [ + { + id: crypto.randomUUID(), + effectiveFrom: new Date().toISOString(), + policy: { kind: "catalog" }, + }, + ], }) .returning({ id: providerModel.id }) const assigned = await transaction diff --git a/platform/src/db/migrations/20261022183633_model-price-history/migration.sql b/platform/src/db/migrations/20261022183633_model-price-history/migration.sql new file mode 100644 index 00000000..2015ff0c --- /dev/null +++ b/platform/src/db/migrations/20261022183633_model-price-history/migration.sql @@ -0,0 +1,7 @@ +ALTER TABLE "chat_message" ADD COLUMN "request_accounting" jsonb;--> statement-breakpoint +ALTER TABLE "provider_model" ADD COLUMN "pricing_history" jsonb DEFAULT '[]' NOT NULL;--> statement-breakpoint +UPDATE provider_model SET pricing_history = jsonb_build_array(jsonb_build_object( + 'id', uuidv7()::text, + 'effectiveFrom', to_char(now() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), + 'rates', jsonb_build_object('currency', configuration->'currency', 'pricingSource', configuration->'pricingSource', 'prices', configuration->'prices', 'contextTiers', configuration->'contextTiers') +)) WHERE configuration->'pricingSource'->>'kind' = 'manual'; diff --git a/platform/src/db/migrations/20261022183633_model-price-history/snapshot.json b/platform/src/db/migrations/20261022183633_model-price-history/snapshot.json new file mode 100644 index 00000000..b7c99613 --- /dev/null +++ b/platform/src/db/migrations/20261022183633_model-price-history/snapshot.json @@ -0,0 +1,81 @@ +{ + "type": "diff", + "version": "9", + "dialect": "postgres", + "id": "54f8935b-b07d-4dec-bc6c-e044ae7cabe7", + "prevIds": [ + "0183fddd-426f-44fa-9670-3fc9cd3249cc" + ], + "footprints": [ + { + "statementType": "add_column", + "statementTarget": [ + "column", + "public", + "chat_message", + "request_accounting" + ], + "writes": [ + "s|col|public.chat_message.request_accounting", + "n|col|public.chat_message.request_accounting", + "e|col|public.chat_message.request_accounting" + ], + "reads": [ + "n|tbl|public.chat_message", + "n|sch|public" + ] + }, + { + "statementType": "add_column", + "statementTarget": [ + "column", + "public", + "provider_model", + "pricing_history" + ], + "writes": [ + "s|col|public.provider_model.pricing_history", + "n|col|public.provider_model.pricing_history", + "e|col|public.provider_model.pricing_history" + ], + "reads": [ + "n|tbl|public.provider_model", + "n|sch|public" + ] + } + ], + "deltas": [ + { + "op": "create", + "row": { + "schema": "public", + "table": "chat_message", + "name": "request_accounting", + "type": "jsonb", + "typeSchema": null, + "notNull": false, + "dimensions": 0, + "default": null, + "generated": null, + "identity": null, + "entityType": "columns" + } + }, + { + "op": "create", + "row": { + "schema": "public", + "table": "provider_model", + "name": "pricing_history", + "type": "jsonb", + "typeSchema": null, + "notNull": true, + "dimensions": 0, + "default": "'[]'", + "generated": null, + "identity": null, + "entityType": "columns" + } + } + ] +} \ No newline at end of file diff --git a/platform/src/db/migrations/20261024151405_simplify-model-pricing/migration.sql b/platform/src/db/migrations/20261024151405_simplify-model-pricing/migration.sql new file mode 100644 index 00000000..f86152ad --- /dev/null +++ b/platform/src/db/migrations/20261024151405_simplify-model-pricing/migration.sql @@ -0,0 +1,15 @@ +UPDATE provider_model SET + pricing_history = CASE WHEN jsonb_array_length(pricing_history) = 0 THEN + jsonb_build_array(jsonb_build_object('id', uuidv7()::text, + 'effectiveFrom', to_char(now() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), + 'policy', CASE WHEN configuration->'pricingSource'->>'kind' = 'manual' + THEN jsonb_build_object('kind', 'manual', 'prices', configuration->'prices') + ELSE jsonb_build_object('kind', 'catalog') END)) + ELSE (SELECT jsonb_agg(jsonb_build_object('id', entry->'id', 'effectiveFrom', entry->'effectiveFrom', + 'policy', CASE WHEN entry->'rates' = 'null'::jsonb THEN jsonb_build_object('kind', 'catalog') + ELSE jsonb_build_object('kind', 'manual', 'prices', entry->'rates'->'prices') END) ORDER BY ordinal) + FROM jsonb_array_elements(pricing_history) WITH ORDINALITY AS changes(entry, ordinal)) END, + configuration = CASE WHEN configuration->'pricingSource'->>'kind' = 'manual' + THEN configuration - 'currency' - 'pricingSource' - 'prices' - 'contextTiers' + ELSE NULL END, + updated_at = now(); diff --git a/platform/src/db/migrations/20261024151405_simplify-model-pricing/snapshot.json b/platform/src/db/migrations/20261024151405_simplify-model-pricing/snapshot.json new file mode 100644 index 00000000..04f37ccc --- /dev/null +++ b/platform/src/db/migrations/20261024151405_simplify-model-pricing/snapshot.json @@ -0,0 +1,11 @@ +{ + "type": "diff", + "version": "9", + "dialect": "postgres", + "id": "ec41487a-0de3-4c96-8338-defe74462cac", + "prevIds": [ + "54f8935b-b07d-4dec-bc6c-e044ae7cabe7" + ], + "footprints": [], + "deltas": [] +} \ No newline at end of file diff --git a/platform/src/db/migrations/20261025053857_request-usage-bounds/migration.sql b/platform/src/db/migrations/20261025053857_request-usage-bounds/migration.sql new file mode 100644 index 00000000..8b9d60d2 --- /dev/null +++ b/platform/src/db/migrations/20261025053857_request-usage-bounds/migration.sql @@ -0,0 +1,9 @@ +ALTER TABLE "chat_message" ADD COLUMN "first_request_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "chat_message" ADD COLUMN "last_request_at" timestamp with time zone;--> statement-breakpoint +UPDATE "chat_message" SET + (first_request_at, last_request_at) = ( + SELECT min((request->>'startedAt')::timestamptz), max((request->>'startedAt')::timestamptz) + FROM jsonb_array_elements(request_accounting->'requests') request + ), updated_at = now() +WHERE request_accounting IS NOT NULL;--> statement-breakpoint +CREATE INDEX "chat_message_user_request_usage_idx" ON "chat_message" ("organization_id","tenant_id",("metadata"->'provenance'->>'initiatorTenantUserId'),"last_request_at","first_request_at") WHERE "role" = 'assistant'; diff --git a/platform/src/db/migrations/20261025053857_request-usage-bounds/snapshot.json b/platform/src/db/migrations/20261025053857_request-usage-bounds/snapshot.json new file mode 100644 index 00000000..d4ac42d1 --- /dev/null +++ b/platform/src/db/migrations/20261025053857_request-usage-bounds/snapshot.json @@ -0,0 +1,156 @@ +{ + "type": "diff", + "version": "9", + "dialect": "postgres", + "id": "737891a8-ca6c-41f1-a4a9-5ecf792954d6", + "prevIds": [ + "ec41487a-0de3-4c96-8338-defe74462cac" + ], + "footprints": [ + { + "statementType": "add_column", + "statementTarget": [ + "column", + "public", + "chat_message", + "first_request_at" + ], + "writes": [ + "s|col|public.chat_message.first_request_at", + "n|col|public.chat_message.first_request_at", + "e|col|public.chat_message.first_request_at" + ], + "reads": [ + "n|tbl|public.chat_message", + "n|sch|public" + ] + }, + { + "statementType": "add_column", + "statementTarget": [ + "column", + "public", + "chat_message", + "last_request_at" + ], + "writes": [ + "s|col|public.chat_message.last_request_at", + "n|col|public.chat_message.last_request_at", + "e|col|public.chat_message.last_request_at" + ], + "reads": [ + "n|tbl|public.chat_message", + "n|sch|public" + ] + }, + { + "statementType": "create_index", + "statementTarget": [ + "index", + "public", + "chat_message", + "chat_message_user_request_usage_idx" + ], + "writes": [ + "s|idx|public.chat_message_user_request_usage_idx", + "n|idx|public.chat_message_user_request_usage_idx", + "e|idx|public.chat_message_user_request_usage_idx", + "n|rel|public.chat_message_user_request_usage_idx" + ], + "reads": [ + "n|tbl|public.chat_message", + "n|sch|public", + "n|col|public.chat_message.organization_id", + "n|col|public.chat_message.tenant_id", + "n|col|public.chat_message.last_request_at", + "n|col|public.chat_message.first_request_at" + ] + } + ], + "deltas": [ + { + "op": "create", + "row": { + "schema": "public", + "table": "chat_message", + "name": "first_request_at", + "type": "timestamp with time zone", + "typeSchema": null, + "notNull": false, + "dimensions": 0, + "default": null, + "generated": null, + "identity": null, + "entityType": "columns" + } + }, + { + "op": "create", + "row": { + "schema": "public", + "table": "chat_message", + "name": "last_request_at", + "type": "timestamp with time zone", + "typeSchema": null, + "notNull": false, + "dimensions": 0, + "default": null, + "generated": null, + "identity": null, + "entityType": "columns" + } + }, + { + "op": "create", + "row": { + "schema": "public", + "table": "chat_message", + "name": "chat_message_user_request_usage_idx", + "nameExplicit": true, + "columns": [ + { + "value": "organization_id", + "isExpression": false, + "asc": true, + "nullsFirst": false, + "opclass": null + }, + { + "value": "tenant_id", + "isExpression": false, + "asc": true, + "nullsFirst": false, + "opclass": null + }, + { + "value": "(\"metadata\"->'provenance'->>'initiatorTenantUserId')", + "isExpression": true, + "asc": true, + "nullsFirst": false, + "opclass": null + }, + { + "value": "last_request_at", + "isExpression": false, + "asc": true, + "nullsFirst": false, + "opclass": null + }, + { + "value": "first_request_at", + "isExpression": false, + "asc": true, + "nullsFirst": false, + "opclass": null + } + ], + "isUnique": false, + "where": "\"role\" = 'assistant'", + "with": "", + "method": "btree", + "concurrently": false, + "entityType": "indexes" + } + } + ] +} \ No newline at end of file diff --git a/platform/src/db/schema/chat.ts b/platform/src/db/schema/chat.ts index 759bf675..7b48c329 100644 --- a/platform/src/db/schema/chat.ts +++ b/platform/src/db/schema/chat.ts @@ -14,11 +14,13 @@ import { } from "drizzle-orm/pg-core" import { ApiUuidSchema } from "@/lib/tenants/schemas.ts" +import { ModelRequestSnapshotSchema } from "@/lib/model-providers/schemas.ts" import { deferrableForeignKey, schemaJsonb, lockVersion, timestamps, + timestampWithTimeZone, uuidV7, } from "../lib/columns.ts" import { fileObject } from "./files.ts" @@ -224,6 +226,14 @@ export const chatMessage = snakeCase.table( // Versioned, validated provenance, provider continuation, and input tool declarations. The current invocation ID fences producer writes. // Credentials are excluded. Visible content belongs in message parts, and provider context is derived separately. metadata: schemaJsonb(ChatMessageMetadataSchema).notNull(), + requestAccounting: schemaJsonb( + Schema.Struct({ + version: Schema.Literals([1, 2]), + requests: Schema.Array(ModelRequestSnapshotSchema), + }).check(boundedChatJson), + ), + firstRequestAt: timestampWithTimeZone(), + lastRequestAt: timestampWithTimeZone(), // Lifecycle on the initiating user message. Its accepted content remains complete throughout execution. turnState: chatMessageTurnStateEnum(), ...timestamps(), @@ -255,6 +265,15 @@ export const chatMessage = snakeCase.table( table.createdAt, ) .where(sql`${table.role} = 'assistant'`), + index("chat_message_user_request_usage_idx") + .on( + table.organizationId, + table.tenantId, + sql`(${table.metadata}->'provenance'->>'initiatorTenantUserId')`, + table.lastRequestAt, + table.firstRequestAt, + ) + .where(sql`${table.role} = 'assistant'`), deferrableForeignKey({ columns: [table.organizationId, table.tenantId, table.threadId], foreignColumns: [chatThread.organizationId, chatThread.tenantId, chatThread.id], diff --git a/platform/src/db/schema/organizations.ts b/platform/src/db/schema/organizations.ts index 84954e85..ddd90f9d 100644 --- a/platform/src/db/schema/organizations.ts +++ b/platform/src/db/schema/organizations.ts @@ -23,7 +23,8 @@ import { } from "@/lib/sandboxes/schemas.ts" import { ModelProviderCredentialsPayloadSchema, - ModelConfigurationSchema, + ModelLimitsSchema, + ModelPriceHistorySchema, type ModelProviderType, type ModelProviderApi, } from "@/lib/model-providers/schemas.ts" @@ -179,7 +180,8 @@ export const providerModel = snakeCase.table( modelProviderId: uuid().notNull(), modelId: text().notNull(), name: text().notNull(), - configuration: schemaJsonb(ModelConfigurationSchema), + configuration: schemaJsonb(ModelLimitsSchema), + pricingHistory: schemaJsonb(ModelPriceHistorySchema).default([]).notNull(), ...timestamps(), }, (table) => [ diff --git a/platform/src/lib/chat/README.md b/platform/src/lib/chat/README.md index 7a17b52b..290578c1 100644 --- a/platform/src/lib/chat/README.md +++ b/platform/src/lib/chat/README.md @@ -178,6 +178,8 @@ Optional fields also preserve input tool declarations, opaque `modelMessages`, p Assistant provenance usage includes a decimal-string `costUsd` when a reported cost or resolved token rate is available. This snapshots the generation-time cost for retained history. Reported usage from failed model calls is retained and recorded once per model phase, including partial Anthropic input/cache usage. One-hour cache writes are priced as a subset of total cache writes. Directory totals exclude drafts and temporarily count unpriceable settled replies as zero. +Assistant messages also retain internal `request_accounting` rate snapshots before provider dispatch. Each attempt has its own ID, UTC start and rate card, including retries and native-web continuations. Received terminal usage updates that attempt's charge using its frozen rates, preferring provider-reported cost. Writer claims fence these writes, and interruption preserves snapshots even without visible output. Version 2 accounting supplies directory totals by request start time. Indexed first/last request timestamps on each message prune unrelated history before the exact per-request time filter. Older messages retain their saved aggregate charge and original message-date attribution. A snapshot alone is not confirmed billable usage. Recovery of usage omitted by providers and reporting incompleteness remain follow-up work. + `initiatorTenantUserId` identifies the actor of this foreground invocation. A participant resolving the last browser response can start a continuation, so that actor can differ from the original user-message author. ### Part payloads diff --git a/platform/src/lib/chat/adapter.ts b/platform/src/lib/chat/adapter.ts index 99d06490..191b0f88 100644 --- a/platform/src/lib/chat/adapter.ts +++ b/platform/src/lib/chat/adapter.ts @@ -6,6 +6,7 @@ import { type AdapterYieldChunk, type TokenUsage, type ContentPart, + fromSpecTokenUsage, } from "@tanstack/ai" import { createAnthropicChat } from "@tanstack/ai-anthropic" import { createGeminiChat } from "@tanstack/ai-gemini" @@ -15,9 +16,11 @@ import { createOpenRouterText } from "@tanstack/ai-openrouter" import { HTTPClient } from "@openrouter/sdk/lib/http" import type { ChatContentItems } from "@openrouter/sdk/models" import { openaiCompatibleText } from "@tanstack/ai-openai/compatible" -import { Effect, Queue, Schema, Stream } from "effect" +import { BigDecimal, Effect, Queue, Schema, Stream } from "effect" import type { ChatModelConfiguration } from "@/lib/model-providers/model-providers.server" +import type { ModelRateCard } from "@/lib/model-providers/schemas" +import { modelRequestCost } from "@/lib/model-providers/usage-cost" import { CHAT_WEB_ACTIVITY_EVENT, fetchChatWeb, type ChatWebState } from "./web" @@ -36,6 +39,7 @@ function chatWebTerminal( terminal: ChatWebTerminal, usages: TokenUsage[], web: ChatWebState, + costs?: string[], ): ChatWebTerminal { web.failed ||= terminal.type === EventType.RUN_ERROR return { @@ -61,7 +65,19 @@ function chatWebTerminal( : [], ), ), - providerUsageDetails: { requestUsage: usages }, + providerUsageDetails: { + requestUsage: usages, + ...(costs + ? { + costUsd: BigDecimal.format( + costs.reduce( + (sum, cost) => BigDecimal.sum(sum, BigDecimal.fromStringUnsafe(cost)), + BigDecimal.fromNumberUnsafe(0), + ), + ), + } + : {}), + }, ...(web.usage.some((usage) => typeof usage.cost === "number") ? { cost: web.usage.reduce((n, usage) => n + Number(usage.cost ?? 0), 0) } : {}), @@ -177,15 +193,36 @@ function createChatProviderAdapter(configuration: ChatModelConfiguration, native }) } -export function createChatAdapter(model: ChatModelConfiguration, web?: ChatWebState) { +export function createChatAdapter( + model: ChatModelConfiguration, + web?: ChatWebState, + accounting?: { + pricing: () => ModelRateCard | null | undefined + recordUsage?: (usage: TokenUsage) => Promise + }, +) { const adapter = createChatProviderAdapter( web ? { ...model, fetch: (input, init) => fetchChatWeb(model, web, input, init) } : model, web !== undefined, ) - if (!web) return adapter + if (!web && !accounting?.recordUsage) return adapter const stream = adapter.chatStream.bind(adapter) as ( options: TextOptions, ) => AsyncIterable + if (!web) { + adapter.chatStream = async function* (options: TextOptions) { + for await (const chunk of stream(options)) { + if (chunk.type === EventType.RUN_FINISHED || chunk.type === EventType.RUN_ERROR) { + const usage = Array.isArray(chunk.usage) + ? fromSpecTokenUsage(chunk.usage, chunk.metadata?.tanstack?.usage) + : (chunk.usage ?? fromSpecTokenUsage(undefined, chunk.metadata?.tanstack?.usage)) + if (usage) await accounting!.recordUsage!(usage) + } + yield chunk + } + } + return adapter + } // Native adapters delay retrieval starts and omit pause_turn and citations. // https://github.com/TanStack/ai/blob/main/packages/ai-anthropic/src/adapters/text.ts adapter.chatStream = async function* (options: TextOptions): AsyncGenerator { @@ -209,6 +246,7 @@ export function createChatAdapter(model: ChatModelConfiguration, web?: ChatWebSt const text = { content: "", messageId: undefined as string | undefined } const signal = options.request?.signal const usages: TokenUsage[] = [] + const costs: string[] = [] let terminal: ChatWebTerminal | undefined const progress = Effect.runSync(Queue.unbounded()) web.onActivity = () => @@ -220,6 +258,7 @@ export function createChatAdapter(model: ChatModelConfiguration, web?: ChatWebSt try { do { web.pause = false + const usageStart = web.usage.length for await (const chunk of Stream.toAsyncIterable( Stream.merge( Stream.fromAsyncIterable(stream({ ...options, messages }), (error) => error), @@ -227,14 +266,29 @@ export function createChatAdapter(model: ChatModelConfiguration, web?: ChatWebSt { haltStrategy: "left" }, ), )) { - const output = chatWebContent(chunk, text, signal) - if (output === null) return - if (!output) continue if (chunk.type === EventType.RUN_FINISHED || chunk.type === EventType.RUN_ERROR) { terminal = chunk - if (chunk.usage && !Array.isArray(chunk.usage)) usages.push(chunk.usage) + if (chunk.usage && !Array.isArray(chunk.usage)) { + usages.push(chunk.usage) + const reportedCost = + web.usage.length > usageStart && Number.isFinite(web.usage.at(-1)?.cost) + ? Number(web.usage.at(-1)!.cost) + : chunk.usage.cost + const usage = { + ...chunk.usage, + ...(reportedCost === undefined ? {} : { cost: reportedCost }), + } + const cost = accounting?.recordUsage + ? await accounting.recordUsage(usage) + : modelRequestCost({ usage, configuration: accounting?.pricing() }) + costs.push(cost ?? "0") + } + if (signal?.aborted) return continue } + const output = chatWebContent(chunk, text, signal) + if (output === null) return + if (!output) continue yield output } if (web.pause && terminal?.type !== EventType.RUN_ERROR) { @@ -254,7 +308,7 @@ export function createChatAdapter(model: ChatModelConfiguration, web?: ChatWebSt } while (web.pause) if (text.messageId) yield { type: EventType.TEXT_MESSAGE_END, messageId: text.messageId } if (terminal) { - yield chatWebTerminal(terminal, usages, web) + yield chatWebTerminal(terminal, usages, web, accounting ? costs : undefined) } } finally { web.onActivity = undefined diff --git a/platform/src/lib/chat/chat.test.ts b/platform/src/lib/chat/chat.test.ts index 3b68c919..7e7f150e 100644 --- a/platform/src/lib/chat/chat.test.ts +++ b/platform/src/lib/chat/chat.test.ts @@ -4,13 +4,15 @@ import { assert, describe, it } from "@effect/vitest" import { chatParamsFromRequestBody, EventType, type ChatMiddleware } from "@tanstack/ai" import { Deferred, Effect, Fiber, Layer, Stream } from "effect" import { beforeEach, vi } from "vitest" +import type { createChatAdapter } from "./adapter" +import type { ModelRequestSnapshot } from "@/lib/model-providers/schemas" const chatRunTest = vi.hoisted(() => ({ options: [] as Array<{ messages: unknown[] systemPrompts: string[] tools: Array<{ name: string }> - adapter: { model: string } + adapter: ReturnType modelOptions?: unknown toolExecution?: string middleware: ChatMiddleware[] @@ -27,7 +29,7 @@ vi.mock("@tanstack/ai", async (original) => ({ messages: unknown[] systemPrompts: string[] tools: Array<{ name: string }> - adapter: { model: string } + adapter: ReturnType abortController: AbortController middleware: ChatMiddleware[] }) => { @@ -108,6 +110,8 @@ function chatTestLayer(options: { readonly history?: MessageRecord[] readonly acquireMcp?: typeof McpTools.Service.acquire readonly sandboxSession?: typeof ChatSandboxes.Service.session + readonly pricing?: typeof ModelProviders.Service.pricing + readonly recordRequest?: typeof ChatThreads.Service.recordRequest }) { const agents = { resolveForChat: () => @@ -136,7 +140,7 @@ function chatTestLayer(options: { test: () => Effect.succeed({ toolCount: 0 }), }), Layer.succeed(ModelProviders, { - pricing: () => Effect.succeed(null), + pricing: options.pricing ?? (() => Effect.succeed(null)), resolveForAgent: () => options.model === undecryptable ? Effect.fail(new ModelProviderUnreadable()) @@ -149,6 +153,7 @@ function chatTestLayer(options: { Layer.succeed(ChatThreads, { history: () => Effect.succeed(options.history ?? []), assertActive: () => Effect.void, + recordRequest: options.recordRequest ?? (() => Effect.void), interrupt: () => Effect.sync(() => chatRunTest.order.push("interrupted")), } as unknown as ChatThreads["Service"]), ]), @@ -187,6 +192,94 @@ beforeEach(() => { }) describe("Chat.run", () => { + it.effect.each(["failed", "cancelled", "charged"])( + "requires a durable request snapshot before dispatch: %s", + (outcome) => { + let recorded = false + let charged: ModelRequestSnapshot | undefined + let dispatched = 0 + const abortController = new AbortController() + return Effect.gen(function* () { + yield* Stream.runCollect(Stream.take(yield* runChat(), 1)) + const adapter = chatRunTest.options[0]!.adapter + yield* Effect.promise(async () => { + const { chat } = await vi.importActual("@tanstack/ai") + for await (const _chunk of chat({ + threadId: "request-snapshot-test", + adapter, + abortController, + messages: [{ role: "user", content: "Hello" }], + })) { + // Drain the native adapter against the synthetic HTTP boundary. + } + }) + assert.strictEqual(dispatched, outcome === "charged" ? 1 : 0) + if (outcome === "charged") { + assert.strictEqual(charged?.costUsd, "0.000036") + assert.strictEqual(charged?.usage?.promptTokens, 10) + assert.strictEqual(charged?.usage?.completionTokens, 2) + } + }).pipe( + Effect.provide( + chatTestLayer({ + agent: { ...sandboxedAgent, sandboxProviderId: null }, + model: { + ...CHAT_TEST_MODEL, + fetch: () => { + assert.strictEqual(recorded, true) + dispatched += 1 + return Promise.resolve( + new Response( + [ + { + type: "response.created", + response: { id: "response", model: "test-model", output: [] }, + }, + { + type: "response.completed", + response: { + id: "response", + model: "test-model", + status: "completed", + output: [], + usage: { input_tokens: 10, output_tokens: 2, total_tokens: 12 }, + }, + }, + ] + .map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`) + .join(""), + { headers: { "Content-Type": "text/event-stream" } }, + ), + ) + }, + }, + pricing: () => + Effect.succeed({ + currency: "USD", + pricingSource: { kind: "manual" }, + prices: { inputPerMillion: "2", outputPerMillion: "8" }, + contextTiers: [], + }), + recordRequest: ({ claim, request }) => { + if (request.usage) charged = request + assert.deepStrictEqual(claim, chatTestExecution.claim) + assert.strictEqual(request.providerId, CHAT_TEST_MODEL.providerId) + assert.strictEqual(request.modelId, CHAT_TEST_MODEL.modelId) + assert.strictEqual(JSON.stringify(request).includes(CHAT_TEST_MODEL.apiKey), false) + return outcome !== "failed" + ? Effect.promise(async () => { + if (outcome === "cancelled") abortController.abort() + await Promise.resolve() + recorded = true + }) + : Effect.die("Snapshot unavailable") + }, + }), + ), + ) + }, + ) + it.effect.each([true, false])( "suppresses tools only for this turn's unknown outcome: %s", (ownTurn) => diff --git a/platform/src/lib/chat/chat.ts b/platform/src/lib/chat/chat.ts index 8fa873e4..34440d92 100644 --- a/platform/src/lib/chat/chat.ts +++ b/platform/src/lib/chat/chat.ts @@ -5,6 +5,7 @@ import { maxIterations, mergeAgentTools, type StreamChunk, + type TokenUsage, } from "@tanstack/ai" import { Context, Effect, identity, Layer, Stream } from "effect" @@ -23,6 +24,9 @@ import { } from "@/lib/model-providers/model-providers.server" import { Agents } from "@/lib/agents/agents.server" import { createChatAdapter } from "./adapter" +import type { ModelRequestSnapshot } from "@/lib/model-providers/schemas" +import { modelRequestCost } from "@/lib/model-providers/usage-cost" +import { chatStoredJson } from "./threads/projection" import { chatWebTools, chatWebState } from "./web" import { createChatAttachmentTools } from "./attachments/tools" import { @@ -254,13 +258,12 @@ export class Chat extends Context.Service< .interrupt({ claim: turn.claim }) .pipe(Effect.timeout("5 seconds"), Effect.ignore), ) - const pricing = yield* modelProviders.pricing(model) const usageLimitGuard = limits.length ? chatUsageLimitGuard({ usageLimits, limits, subject: { ...input.managed.claim.scope, threadId: input.managed.claim.threadId }, - pricing, + pricing: () => managed.state.pricing, execute: Effect.runPromiseWith(services), }) : undefined @@ -324,7 +327,6 @@ export class Chat extends Context.Service< history: convertMessagesToModelMessages(messages), tools, model, - pricing, web, publicHistory: web || @@ -392,8 +394,62 @@ export class Chat extends Context.Service< ...(usageLimitGuard ? [usageLimitGuard.middleware] : []), ] return chatEventStream((abortController) => { + let requestSnapshot: ModelRequestSnapshot | undefined + const recordUsage = async (usage: TokenUsage) => { + if (!requestSnapshot) return null + requestSnapshot = { + ...requestSnapshot, + usage: chatStoredJson(usage), + costUsd: modelRequestCost({ usage, configuration: requestSnapshot.rates }), + } + await Effect.runPromiseWith(services)( + threads + .recordRequest({ claim: turn.claim, request: requestSnapshot }) + .pipe(Effect.timeout("5 seconds")), + ) + return requestSnapshot.costUsd ?? null + } const source = chat({ - adapter: createChatAdapter(model, web), + adapter: createChatAdapter( + { + ...model, + fetch: async (request, init) => { + const signal = + init?.signal ?? (request instanceof Request ? request.signal : undefined) + signal?.throwIfAborted() + const timestamp = new Date() + const rates = await Effect.runPromiseWith(services)( + modelProviders.pricing({ ...model, timestamp }), + signal ? { signal } : undefined, + ) + managed.state.pricing = rates + requestSnapshot = { + id: crypto.randomUUID(), + startedAt: timestamp.toISOString(), + providerId: model.providerId, + providerType: model.providerType, + protocol: model.api, + modelId: model.modelId, + rates: rates + ? { + currency: rates.currency, + pricingSource: rates.pricingSource, + prices: rates.prices, + contextTiers: rates.contextTiers, + } + : null, + } + await Effect.runPromiseWith(services)( + threads.recordRequest({ claim: turn.claim, request: requestSnapshot }), + signal ? { signal } : undefined, + ) + signal?.throwIfAborted() + return model.fetch(request, init) + }, + }, + web, + { pricing: () => managed.state.pricing, recordUsage }, + ), messages, systemPrompts, // Host tools arrive declared in the request body and run in the page. `mergeAgentTools` diff --git a/platform/src/lib/chat/threads/stream.test.ts b/platform/src/lib/chat/threads/stream.test.ts index 01ba70ec..cd0bf5cd 100644 --- a/platform/src/lib/chat/threads/stream.test.ts +++ b/platform/src/lib/chat/threads/stream.test.ts @@ -174,10 +174,6 @@ async function exerciseManagedStream(options: { pricingSource: { kind: "manual" }, prices: { inputPerMillion: "3", outputPerMillion: "15" }, contextTiers: [], - maxInputTokens: 100_000, - contextWindowTokens: null, - maxOutputTokens: 4000, - outputCap: 1000, }, agentId: "agent", execute: Effect.runPromise, diff --git a/platform/src/lib/chat/threads/stream.ts b/platform/src/lib/chat/threads/stream.ts index 0c64db9f..7d1d23c3 100644 --- a/platform/src/lib/chat/threads/stream.ts +++ b/platform/src/lib/chat/threads/stream.ts @@ -17,13 +17,13 @@ import { PersistenceCompletionCapability, withPersistence, } from "@tanstack/ai-persistence" -import { BigDecimal, Effect, Schema } from "effect" +import { Effect, Schema } from "effect" import type { ChatFileCache } from "../attachments/attachments" import { APP_HANDLE } from "@/lib/constants" import type { ChatModelConfiguration } from "@/lib/model-providers/model-providers.server" -import type { ModelConfiguration } from "@/lib/model-providers/schemas" -import { modelUsageCost } from "@/lib/model-providers/usage-cost" +import type { ModelRateCard } from "@/lib/model-providers/schemas" +import { modelRequestCost } from "@/lib/model-providers/usage-cost" import type { ChatThreads } from "./threads" import type { ChatMessagePayload, ChatWriterClaim } from "./schemas" import { chatStoredJson, chatPublicParts } from "./projection" @@ -51,7 +51,7 @@ interface ManagedChatStreamOptions { metadata?: Readonly> | undefined }> readonly model: ChatModelConfiguration - readonly pricing?: ModelConfiguration | null | undefined + readonly pricing?: ModelRateCard | null | undefined readonly agentId: string readonly web?: ChatWebState | undefined readonly publicHistory?: UIMessage[] | undefined @@ -77,6 +77,7 @@ interface ManagedChatStreamState { version: number payload: ChatMessagePayload usage: typeof Schema.JsonObject.Type | undefined + pricing: ModelRateCard | null | undefined publicHistory: UIMessage[] | undefined readonly fileCache: ChatFileCache readonly partIds: Map @@ -386,6 +387,7 @@ export function managedChatMiddleware(options: ManagedChatStreamOptions) { version: options.managed.threadVersion, payload: { version: 1, parts: [] }, usage: undefined, + pricing: options.pricing, publicHistory: options.publicHistory, fileCache: new Map(), partIds: new Map(), @@ -479,12 +481,7 @@ export function managedChatMiddleware(options: ManagedChatStreamOptions) { onUsage(_ctx, usage) { state.usage = chatStoredJson({ ...usage, - costUsd: - typeof usage.cost === "number" && Number.isFinite(usage.cost) - ? BigDecimal.format(BigDecimal.fromNumberUnsafe(usage.cost)) - : options.pricing - ? modelUsageCost(options.pricing, usage) - : null, + costUsd: modelRequestCost({ usage, configuration: state.pricing }), }) }, async onInterruptBoundary(ctx) { diff --git a/platform/src/lib/chat/threads/threads.integration.test.ts b/platform/src/lib/chat/threads/threads.integration.test.ts index d2c78391..ae9ff52d 100644 --- a/platform/src/lib/chat/threads/threads.integration.test.ts +++ b/platform/src/lib/chat/threads/threads.integration.test.ts @@ -41,6 +41,8 @@ import { tenantUser, } from "@/db/schema" import { ChatThreads } from "./threads" +import { TenantUsers } from "@/lib/tenants/tenant-users" +import type { ModelRequestSnapshot } from "@/lib/model-providers/schemas" import { projectChatModelHistory } from "./projection" import type { ChatThreadScope, @@ -88,7 +90,9 @@ const filesLayer = ChatFiles.layerNoDeps.pipe( Layer.provideMerge(StoredFiles.layerNoDeps.pipe(Layer.provideMerge(storage))), Layer.provideMerge(Database.layer), ) -const runtime = ManagedRuntime.make(ChatThreads.layerNoDeps.pipe(Layer.provideMerge(filesLayer))) +const runtime = ManagedRuntime.make( + Layer.mergeAll(ChatThreads.layerNoDeps.pipe(Layer.provideMerge(filesLayer)), TenantUsers.layer), +) describe.skipIf(!integration.url)("PostgreSQL chat conversations", () => { let db: ReturnType @@ -153,6 +157,142 @@ describe.skipIf(!integration.url)("PostgreSQL chat conversations", () => { }), ) + test("request snapshots are immutable, writer-scoped and retained without visible output", async () => { + const thread = await create() + const admitted = await admit(thread.id) + const claim = admitted.claim! + const request: ModelRequestSnapshot = { + id: crypto.randomUUID(), + startedAt: "2026-10-22T12:00:00.000Z", + providerId: crypto.randomUUID(), + providerType: "openai", + protocol: "responses", + modelId: "test-model", + rates: { + currency: "USD", + pricingSource: { kind: "manual" }, + prices: { inputPerMillion: "2", outputPerMillion: "8" }, + contextTiers: [], + }, + } + await runtime.runPromise(service.recordRequest({ claim, request })) + await runtime.runPromise(service.recordRequest({ claim, request })) + const retry = { + ...request, + id: crypto.randomUUID(), + rates: null, + startedAt: "2026-11-01T00:00:00.000Z", + } + await runtime.runPromise(service.recordRequest({ claim, request: retry })) + const earlier = { ...request, id: crypto.randomUUID(), startedAt: "2026-09-30T23:59:59.999Z" } + await runtime.runPromise(service.recordRequest({ claim, request: earlier })) + await expect( + runtime.runPromise(service.recordRequest({ claim, request: { ...request, rates: null } })), + ).rejects.toMatchObject({ _tag: "ChatThreadConflict" }) + for (const unauthorized of [ + other, + foreign, + { ...scope, organizationId: crypto.randomUUID() }, + ]) { + await expect( + runtime.runPromise( + service.recordRequest({ claim: { ...claim, scope: unauthorized }, request }), + ), + ).rejects.toBeDefined() + } + const recorded = { + ...request, + usage: { promptTokens: 10, completionTokens: 1, cost: 1.25 }, + costUsd: "1.25", + } + await runtime.runPromise(service.recordRequest({ claim, request: recorded })) + const charged = { + ...recorded, + usage: { promptTokens: 20, completionTokens: 2, cost: 2.5 }, + costUsd: "2.5", + } + const retried = { + ...retry, + usage: { promptTokens: 10, completionTokens: 1, cost: 3.75 }, + costUsd: "3.75", + } + await runtime.runPromise(service.recordRequest({ claim, request: charged })) + await runtime.runPromise(service.recordRequest({ claim, request: charged })) + await runtime.runPromise(service.recordRequest({ claim, request: retried })) + const users = await runtime.runPromise(TenantUsers) + const query = { + scope: { organizationId: scope.organizationId }, + tenantId: scope.tenantId, + userIds: [scope.tenantUserId, other.tenantUserId], + start: "2026-10-01T00:00:00.000Z", + end: "2026-11-01T00:00:00.000Z", + } + expect( + (await runtime.runPromise(users.usage(query))).items.every(({ costUsd }) => costUsd === "0"), + ).toBe(true) + await runtime.runPromise(service.checkpoint({ claim, payload, state: "complete" })) + await expect( + runtime.runPromise( + service.recordRequest({ claim, request: { ...request, id: crypto.randomUUID() } }), + ), + ).rejects.toMatchObject({ _tag: "ChatThreadConflict" }) + const next = await runtime.runPromise(service.nextDraft({ claim })) + const second = { ...request, id: crypto.randomUUID(), rates: null } + await runtime.runPromise(service.recordRequest({ claim: next, request: second })) + await runtime.runPromise(service.interrupt({ claim: next })) + await expect( + runtime.runPromise(service.recordRequest({ claim: next, request: second })), + ).rejects.toMatchObject({ _tag: "ChatThreadConflict" }) + const history = await runtime.runPromise(service.history({ scope, id: thread.id })) + const settled = history.find((message) => message.id === claim.assistantMessageId)! + const interrupted = history.find((message) => message.id === next.assistantMessageId)! + expect(settled.requestAccounting).toEqual({ version: 2, requests: [charged, retried, earlier] }) + expect(settled).toMatchObject({ + firstRequestAt: new Date(earlier.startedAt), + lastRequestAt: new Date(retry.startedAt), + }) + expect(interrupted).toMatchObject({ + state: "interrupted", + payload: { parts: [] }, + requestAccounting: { version: 2, requests: [second] }, + }) + expect((await runtime.runPromise(users.usage(query))).items).toEqual( + expect.arrayContaining([ + { tenantUserId: scope.tenantUserId, costUsd: "2.5" }, + { tenantUserId: other.tenantUserId, costUsd: "0" }, + ]), + ) + expect( + ( + await runtime.runPromise( + users.usage({ ...query, start: query.end, end: "2026-12-01T00:00:00.000Z" }), + ) + ).items.find(({ tenantUserId }) => tenantUserId === scope.tenantUserId)?.costUsd, + ).toBe("3.75") + await expect( + runtime.runPromise(users.usage({ ...query, scope: { organizationId: crypto.randomUUID() } })), + ).rejects.toMatchObject({ _tag: "TenantNotFound" }) + await db.execute(sql`update chat_message set + request_accounting = jsonb_set(request_accounting, '{version}', '1'), + metadata = jsonb_set(metadata, '{provenance,usage}', '{"costUsd":"9.99"}'), + created_at = '2026-10-22T12:00:00Z', updated_at = now() + where organization_id = ${scope.organizationId} and tenant_id = ${scope.tenantId} and thread_id = ${thread.id} and id = ${claim.assistantMessageId}`) + expect( + (await runtime.runPromise(users.usage(query))).items.find( + ({ tenantUserId }) => tenantUserId === scope.tenantUserId, + )?.costUsd, + ).toBe("9.99") + expect( + ( + await runtime.runPromise( + users.usage({ ...query, start: query.end, end: "2026-12-01T00:00:00.000Z" }), + ) + ).items.every(({ costUsd }) => costUsd === "0"), + ).toBe(true) + expect(messageResource(settled)).not.toHaveProperty("requestAccounting") + expect(messageResource(interrupted)).not.toHaveProperty("requestAccounting") + }) + test("published files belong to the assistant message, remain scoped, and retire with their conversation", async () => { const thread = await create() const { claim } = await admit(thread.id) diff --git a/platform/src/lib/chat/threads/threads.ts b/platform/src/lib/chat/threads/threads.ts index e5383e04..222f32ed 100644 --- a/platform/src/lib/chat/threads/threads.ts +++ b/platform/src/lib/chat/threads/threads.ts @@ -40,6 +40,7 @@ import { CHAT_SANDBOX_MAX_ARTIFACT_BYTES } from "../sandbox/constants" import { fileObject } from "@/db/schema/files" import { CHAT_ATTACHMENT_MAX_TOTAL_BYTES } from "../attachments/constants" import { parseAgentId } from "@/lib/agents/schemas" +import type { ModelRequestSnapshot } from "@/lib/model-providers/schemas" import { tenantUserEmail } from "@/lib/tenants/schemas" import type { TenantScope } from "@/lib/tenants/tenants" import { tenantSearchPattern } from "@/lib/tenants/tenants" @@ -1015,6 +1016,10 @@ export class ChatThreads extends Context.Service< readonly assertActive: (input: { claim: ChatWriterClaim }) => Effect.Effect + readonly recordRequest: (input: { + claim: ChatWriterClaim + request: ModelRequestSnapshot + }) => Effect.Effect readonly publishArtifact: (input: { claim: ChatWriterClaim bytes: Uint8Array @@ -1696,6 +1701,52 @@ export class ChatThreads extends Context.Service< .pipe(mapDatabaseErrors()) }) + const recordRequest = Effect.fn("ChatThreads.recordRequest")( + (input: { claim: ChatWriterClaim; request: ModelRequestSnapshot }) => + db + .transaction((tx) => + Effect.gen(function* () { + const { message } = yield* checkClaim(tx, input.claim) + if (message.state !== "draft") return yield* new ChatThreadConflict() + const requests = message.requestAccounting?.requests ?? [] + const existing = requests.find((request) => request.id === input.request.id) + if (existing) { + const { usage: _usage, costUsd: _cost, ...snapshot } = existing + const { usage: _nextUsage, costUsd: _nextCost, ...nextSnapshot } = input.request + if (!isDeepStrictEqual(snapshot, nextSnapshot)) + return yield* new ChatThreadConflict() + if ( + input.request.usage === undefined || + isDeepStrictEqual(existing, input.request) + ) + return + } + yield* tx + .update(chatMessage) + .set({ + firstRequestAt: sql`least(${chatMessage.firstRequestAt}, ${input.request.startedAt}::timestamptz)`, + lastRequestAt: sql`greatest(${chatMessage.lastRequestAt}, ${input.request.startedAt}::timestamptz)`, + requestAccounting: { + version: 2, + requests: existing + ? requests.map((request) => + request.id === input.request.id ? input.request : request, + ) + : [...requests, input.request], + }, + }) + .where( + and( + messageWhere(input.claim.scope, input.claim.threadId), + eq(chatMessage.id, input.claim.assistantMessageId), + eq(chatMessage.state, "draft"), + ), + ) + }), + ) + .pipe(mapDatabaseErrors()), + ) + const assertActive = Effect.fn("ChatThreads.assertActive")( ({ claim }: { claim: ChatWriterClaim }) => db @@ -1904,6 +1955,7 @@ export class ChatThreads extends Context.Service< removeParticipant, admit, checkpoint, + recordRequest, assertActive, publishArtifact, nextDraft, diff --git a/platform/src/lib/chat/usage-limits.test.ts b/platform/src/lib/chat/usage-limits.test.ts index 4d270612..da8223e3 100644 --- a/platform/src/lib/chat/usage-limits.test.ts +++ b/platform/src/lib/chat/usage-limits.test.ts @@ -36,7 +36,7 @@ it("stops before the next model call and announces the limit once before the run usageLimits, limits: [{} as UsageLimit], subject: { organizationId: "org", tenantId: "tenant", tenantUserId: "user" }, - pricing: null, + pricing: () => null, execute: Effect.runPromise, }) // Managed persistence holds the finish until its delivery ends the stream. diff --git a/platform/src/lib/chat/usage-limits.ts b/platform/src/lib/chat/usage-limits.ts index 52242d2c..1de2ad56 100644 --- a/platform/src/lib/chat/usage-limits.ts +++ b/platform/src/lib/chat/usage-limits.ts @@ -2,8 +2,8 @@ import { fromSpecTokenUsage, EventType, type ChatMiddleware, type StreamChunk } import { Effect, Exit } from "effect" import { APP_HANDLE } from "@/lib/constants" -import type { ModelConfiguration } from "@/lib/model-providers/schemas" -import { modelUsageCost } from "@/lib/model-providers/usage-cost" +import type { ModelRateCard } from "@/lib/model-providers/schemas" +import { modelRequestCost } from "@/lib/model-providers/usage-cost" import { reportFailure } from "@/lib/runtime/failure-report.server" import type { UsageSubject } from "@/lib/usage-limits/counters" import type { UsageLimitReached } from "@/lib/usage-limits/errors" @@ -20,7 +20,7 @@ export function chatUsageLimitGuard(options: { readonly usageLimits: UsageLimits["Service"] readonly limits: readonly UsageLimit[] readonly subject: UsageSubject - readonly pricing: ModelConfiguration | null + readonly pricing: () => ModelRateCard | null | undefined readonly execute: (effect: Effect.Effect) => Promise }) { const { usageLimits, limits, subject } = options @@ -28,7 +28,7 @@ export function chatUsageLimitGuard(options: { const middleware: ChatMiddleware = { name: "usage-limits", async onUsage(_ctx, usage) { - const spend = options.pricing ? modelUsageCost(options.pricing, usage) : "0" + const spend = modelRequestCost({ usage, configuration: options.pricing() }) ?? "0" const tokens = Math.max(usage.totalTokens, usage.promptTokens + usage.completionTokens) await options.execute( usageLimits.record({ limits, subject, usage: { tokens, spend } }).pipe( diff --git a/platform/src/lib/chat/web.test.ts b/platform/src/lib/chat/web.test.ts index 3f15772a..9d347bbc 100644 --- a/platform/src/lib/chat/web.test.ts +++ b/platform/src/lib/chat/web.test.ts @@ -9,6 +9,8 @@ import { import { Effect, Schema } from "effect" import { expect, test } from "vitest" import type { ChatModelConfiguration } from "@/lib/model-providers/model-providers.server" +import type { ModelRateCard } from "@/lib/model-providers/schemas" +import { modelRequestCost } from "@/lib/model-providers/usage-cost" import { createChatAdapter } from "./adapter" import { chatWebState, chatWebTools, fetchChatWeb } from "./web" import { modelWebAccess } from "@/lib/model-providers/web.server" @@ -257,6 +259,7 @@ async function exerciseWeb( cancel?: boolean reject?: boolean slow?: boolean + pricing?: (request: number) => ModelRateCard } = {}, ) { const requests: Schema.JsonObject[] = [] @@ -317,6 +320,7 @@ async function exerciseWeb( } const web = chatWebState() const saved: ChatMessagePayload[] = [] + const requestCosts: (string | null)[] = [] let results = 0 const claim: ChatWriterClaim = { scope: { @@ -441,7 +445,23 @@ async function exerciseWeb( managed, state: bridge.state, source: chat({ - adapter: createChatAdapter(model, web), + adapter: createChatAdapter( + model, + web, + options.pricing + ? { + pricing: () => options.pricing!(requests.length), + recordUsage: (usage) => { + const cost = modelRequestCost({ + usage, + configuration: options.pricing!(requests.length), + }) + requestCosts.push(cost) + return Promise.resolve(cost) + }, + } + : undefined, + ), messages: history, tools, threadId: claim.threadId, @@ -456,7 +476,7 @@ async function exerciseWeb( } catch (error) { if (!options.cancel && !options.reject && options.pause !== "always") throw error } - return { model, web, saved, chunks, requests, signals, results } + return { model, web, saved, chunks, requests, signals, results, requestCosts } } test.each(["openai", "anthropic"] as const)( @@ -662,7 +682,15 @@ test("native errors and cancellation retain observed evidence without applicatio test.each(["once", "always"] as const)( "Anthropic %s pause preserves context and stays within the turn budget", async (pause) => { - const result = await exerciseWeb("anthropic", { pause }) + const result = await exerciseWeb("anthropic", { + pause, + pricing: (request) => ({ + currency: "USD", + pricingSource: { kind: "manual" }, + prices: { inputPerMillion: request === 1 ? "2" : "4", outputPerMillion: "8" }, + contextTiers: [], + }), + }) expect(result.requests).toHaveLength(pause === "once" ? 2 : CHAT_MAX_MODEL_TURNS) expect(JSON.stringify(result.requests[1])).toContain("web_fetch_result") const usage = Schema.decodeUnknownSync(Schema.UndefinedOr(Schema.JsonObject))( @@ -672,6 +700,8 @@ test.each(["once", "always"] as const)( pause === "once" ? [24, 10, 34] : [undefined, undefined, undefined], ) expect(usage?.promptTokensDetails).toEqual(pause === "once" ? { cachedTokens: 4 } : undefined) + expect(usage?.costUsd).toBe(pause === "once" ? "0.000152" : undefined) + expect(result.requestCosts.slice(0, 2)).toEqual(["0.000064", "0.000088"]) expect(result.chunks.some((chunk) => chunk.type === EventType.RUN_ERROR)).toBe( pause === "always", ) diff --git a/platform/src/lib/model-providers/model-providers.integration.test.ts b/platform/src/lib/model-providers/model-providers.integration.test.ts index dec3e20b..0f2e6da3 100644 --- a/platform/src/lib/model-providers/model-providers.integration.test.ts +++ b/platform/src/lib/model-providers/model-providers.integration.test.ts @@ -34,20 +34,23 @@ import { readModelPriceCatalog, writeModelPriceCatalog, } from "./pricing-catalog.server.ts" -import type { ModelConfiguration } from "./schemas.ts" import modelPriceCatalogRefresh from "../workflows/model-price-catalog-refresh.ts" import { ModelProviders, type SaveModelProviderInput } from "./model-providers.server.ts" -const modelTestConfiguration: ModelConfiguration = { - currency: "USD", - pricingSource: { kind: "manual" }, - prices: { inputPerMillion: "2", outputPerMillion: "8" }, - contextTiers: [], - maxInputTokens: 128_000, - contextWindowTokens: null, - maxOutputTokens: 8192, - outputCap: 4096, -} +const modelTestFields = { + modelId: "same-model", + name: "Shared model name", + configuration: { + maxInputTokens: 128_000, + contextWindowTokens: null, + maxOutputTokens: 8192, + outputCap: 4096, + }, + pricing: { + expectedVersionId: null, + policy: { kind: "manual", prices: { inputPerMillion: "2", outputPerMillion: "8" } }, + }, +} satisfies SaveModelProviderInput["models"][number] const modelIntegrationKey = `sk-${"x".repeat(32)}` @@ -65,13 +68,7 @@ function saveIntegrationProvider( api: "responses", baseUrl: "https://api.openai.com/v1", apiKey: modelIntegrationKey, - models: [ - { - modelId: "same-model", - configuration: modelTestConfiguration, - name: "Shared model name", - }, - ], + models: [modelTestFields], ...fields, }), ) @@ -160,6 +157,173 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () expect(refused._tag).toBe("ModelProviderUnreadable") }) + test("reads effective rates afresh for each request", async () => { + const id = await runAppEffect(saveIntegrationProvider(organizationId)) + const service = await runAppEffect(ModelProviders) + const saved = (await runAppEffect(service.get({ organizationId, id })))! + await db + .insert(agentModel) + .values({ organizationId, agentId, providerModelId: saved.models[0]!.id, position: 0 }) + const model = (await runAppEffect(service.resolveForAgent({ organizationId, agentId })))! + expect(await runAppEffect(service.pricing({ ...model, timestamp: new Date(0) }))).toBeNull() + const future = { + id: crypto.randomUUID(), + effectiveFrom: new Date(Date.now() + 86_400_000).toISOString(), + policy: { kind: "manual" as const, prices: { inputPerMillion: "3", outputPerMillion: "12" } }, + } + await runAppEffect( + saveIntegrationProvider(organizationId, { + id, + lockVersion: 0, + models: [ + { + ...modelTestFields, + pricing: { + expectedVersionId: saved.models[0]!.priceVersionId, + scheduledPrices: [future], + }, + }, + ], + }), + ) + const before = await runAppEffect( + service.pricing({ + ...model, + timestamp: new Date(new Date(future.effectiveFrom).getTime() - 1), + }), + ) + expect(before!.prices).toEqual(modelTestFields.pricing.policy.prices) + expect( + (await runAppEffect( + service.pricing({ ...model, timestamp: new Date(future.effectiveFrom) }), + ))!.prices, + ).toEqual(future.policy.prices) + await runAppEffect( + saveIntegrationProvider(organizationId, { + id, + lockVersion: 1, + models: [ + { + ...modelTestFields, + pricing: { + expectedVersionId: saved.models[0]!.priceVersionId, + policy: { kind: "manual", prices: { inputPerMillion: "4", outputPerMillion: "16" } }, + scheduledPrices: [], + }, + }, + ], + }), + ) + expect((await runAppEffect(service.pricing(model)))!.prices.inputPerMillion).toBe("4") + expect( + await runAppEffect( + service.pricing({ + ...model, + pricingModel: { ...model.pricingModel!, organizationId: crypto.randomUUID() }, + }), + ), + ).toBeNull() + }) + + test("rejects cancelling a scheduled price that activates while a save waits for the provider lock", async () => { + const effectiveFrom = new Date(Date.now() + 86_400_000).toISOString() + const id = await runAppEffect( + saveIntegrationProvider(organizationId, { + models: [ + { + ...modelTestFields, + pricing: { + ...modelTestFields.pricing, + scheduledPrices: [ + { + id: crypto.randomUUID(), + effectiveFrom, + policy: { + kind: "manual", + prices: { inputPerMillion: "3", outputPerMillion: "12" }, + }, + }, + ], + }, + }, + ], + }), + ) + const service = await runAppEffect(ModelProviders) + const saved = (await runAppEffect(service.get({ organizationId, id })))! + const locked = Promise.withResolvers() + const release = Promise.withResolvers() + const blocker = db.transaction(async (transaction) => { + await transaction.execute( + sql`select id from model_provider where organization_id = ${organizationId} and id = ${id} for update`, + ) + locked.resolve() + await release.promise + }) + await locked.promise + const saving = runAppEffect( + service + .save({ + ...saved, + organizationId, + apiKey: null, + models: [ + { + ...modelTestFields, + pricing: { + expectedVersionId: saved.models[0]!.priceVersionId, + policy: { kind: "manual", prices: modelTestFields.pricing.policy.prices }, + scheduledPrices: [], + }, + }, + ], + }) + .pipe(Effect.result), + ) + try { + await vi.waitFor(async () => { + const waiting = await db + .select({ pid: sql`pid` }) + .from(sql`pg_stat_activity`) + .where( + sql`datname = current_database() and wait_event_type = 'Lock' and query ilike 'update "model_provider"%'`, + ) + expect(waiting).toHaveLength(1) + }) + vi.useFakeTimers({ toFake: ["Date"] }) + vi.setSystemTime(new Date(effectiveFrom)) + release.resolve() + await blocker + expect(await saving).toMatchObject({ + _tag: "Failure", + failure: { _tag: "ModelProviderChanged" }, + }) + const current = (await runAppEffect(service.get({ organizationId, id })))! + expect(current.lockVersion).toBe(saved.lockVersion) + expect(current.models[0]!.configuration!.prices.inputPerMillion).toBe("3") + await runAppEffect( + service.save({ + ...saved, + organizationId, + name: "Renamed after activation", + apiKey: null, + models: current.models.map((model) => ({ + modelId: model.modelId, + name: model.name, + configuration: model.limitOverrides, + })), + }), + ) + const renamed = (await runAppEffect(service.get({ organizationId, id })))! + expect(renamed.models[0]!.priceVersionId).toBe(current.models[0]!.priceVersionId) + } finally { + release.resolve() + await blocker + await saving + vi.useRealTimers() + } + }) + test("persists normalized catalog data and retains it after an invalid refresh", async () => { const untrusted = await runAppEffect( saveIntegrationProvider(organizationId, { @@ -167,10 +331,6 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () { modelId: "unknown", name: "Unknown", - configuration: { - ...modelTestConfiguration, - pricingSource: { kind: "catalog", modelId: "unknown" }, - }, }, ], }).pipe(Effect.flip), @@ -245,10 +405,6 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () { modelId: "test-model", name: "Catalog model", - configuration: { - ...modelTestConfiguration, - pricingSource: { kind: "catalog", modelId: "test-model" }, - }, }, ], }), @@ -258,10 +414,7 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () service.get({ organizationId, id: providerId }), ), ) - expect(retained!.models[0]!.configuration).toMatchObject({ - pricingSource: { kind: "catalog" }, - prices: { inputPerMillion: "3", outputPerMillion: "4" }, - }) + expect(retained!.models[0]!.configuration).toBeNull() const cancelDownload = vi.fn() for (const response of [ @@ -379,6 +532,11 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () lockVersion: 0, name: "Renamed", apiKey: null, + models: provider!.models.map(({ modelId, name, limitOverrides }) => ({ + modelId, + name, + configuration: limitOverrides, + })), }), ) const stale = await runAppEffect( @@ -510,8 +668,8 @@ describe.skipIf(!modelProviderIntegration.url)("model provider persistence", () const providerId = await runAppEffect( saveIntegrationProvider(organizationId, { models: [ - { modelId: "first", configuration: modelTestConfiguration, name: "First" }, - { modelId: "second", configuration: modelTestConfiguration, name: "Second" }, + { ...modelTestFields, modelId: "first", name: "First" }, + { ...modelTestFields, modelId: "second", name: "Second" }, ], }), ) diff --git a/platform/src/lib/model-providers/model-providers.server.ts b/platform/src/lib/model-providers/model-providers.server.ts index bc0548bf..35100986 100644 --- a/platform/src/lib/model-providers/model-providers.server.ts +++ b/platform/src/lib/model-providers/model-providers.server.ts @@ -28,15 +28,24 @@ import { ModelProviderTestRateLimited, ModelConfigurationMissing, } from "./errors.ts" -import { effectiveModelConfiguration, readModelPriceCatalog } from "./pricing-catalog.server.ts" +import { + catalogModelConfiguration, + effectiveModelConfiguration, + readModelPriceCatalog, +} from "./pricing-catalog.server.ts" +import { activeModelPrice, updateModelPriceHistory } from "./pricing-history.server.ts" import { testProviderModel } from "./test-model.server.ts" import { modelWebAccess } from "./web.server" import { ModelProviderCredentialsPayloadSchema, - type ModelProviderFields, + type ModelProviderWriteFields, type ModelProviderApi, type ModelProviderType, type ModelConfiguration, + type ModelLimits, + type ModelRateCard, + type ModelPriceVersion, + type ModelScheduledPrice, } from "./schemas.ts" interface ModelProviderModel { @@ -44,6 +53,9 @@ interface ModelProviderModel { readonly modelId: string readonly name: string readonly configuration: ModelConfiguration | null + readonly limitOverrides: ModelLimits | null + readonly priceVersionId: string | null + readonly scheduledPrices: readonly ModelScheduledPrice[] } export interface OrganizationModelProvider { @@ -63,7 +75,7 @@ export interface OrganizationModelProvider { export type ModelProviderListItem = Omit -export interface ModelChoice extends Omit { +export interface ModelChoice extends Pick { readonly providerId: string readonly providerName: string readonly webAccess: boolean | null @@ -77,13 +89,12 @@ export interface ChatModelConfiguration { readonly baseUrl: string readonly apiKey: string readonly modelId: string - /** The model's stored price and limit override, which `ModelProviders.pricing` completes. */ - readonly configured?: ModelConfiguration | null | undefined + readonly pricingModel?: { readonly organizationId: string; readonly id: string } | undefined /** Enforces the deployment's private endpoint policy on every provider request. */ readonly fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise } -export type SaveModelProviderInput = ModelProviderFields & { +export type SaveModelProviderInput = ModelProviderWriteFields & { readonly organizationId: string readonly id: string | null readonly lockVersion: number | null @@ -96,6 +107,13 @@ type TestModelProviderInput = { readonly modelId: string } +type ModelPricingInput = Pick< + ChatModelConfiguration, + "providerType" | "modelId" | "pricingModel" +> & { + readonly timestamp?: Date | undefined +} + type ModelProviderWriteError = | ModelProviderChanged | ModelProviderEndpointNotAllowed @@ -175,9 +193,7 @@ export class ModelProviders extends Context.Service< readonly agentId: string }) => Effect.Effect /** A resolved model's effective rates, reading the shared catalog only without a manual override. */ - readonly pricing: ( - model: Pick, - ) => Effect.Effect + readonly pricing: (model: ModelPricingInput) => Effect.Effect readonly testModel: ( input: TestModelProviderInput, ) => Effect.Effect< @@ -253,6 +269,7 @@ export class ModelProviders extends Context.Service< const list = Effect.fn("ModelProviders.list")(function* (input: { organizationId: string }) { const catalog = yield* readPricing + const timestamp = new Date() const rows = yield* db .select(modelProviderReadColumns) .from(modelProvider) @@ -270,15 +287,25 @@ export class ModelProviders extends Context.Service< credentialsReadable: apiKey !== null, models: models .filter((model) => model.modelProviderId === row.id) - .map(({ id, modelId, name, configuration }) => ({ + .map(({ id, modelId, name, configuration, pricingHistory }) => ({ id, modelId, name, + limitOverrides: configuration, + priceVersionId: + activeModelPrice({ history: pricingHistory, timestamp })?.id ?? null, + scheduledPrices: pricingHistory.filter( + (version): version is ModelScheduledPrice => + version.effectiveFrom > timestamp.toISOString() && + version.policy.kind === "manual", + ), configuration: effectiveModelConfiguration({ catalog, providerType: row.providerType, modelId, configured: configuration, + history: pricingHistory, + timestamp, }), })), } @@ -293,6 +320,7 @@ export class ModelProviders extends Context.Service< if (!stored) return null const { storedCredentials: _storedCredentials, ...row } = stored const catalog = yield* readPricing + const timestamp = new Date() const apiKey = readModelProviderKey(stored) const models = yield* db .select({ @@ -300,6 +328,7 @@ export class ModelProviders extends Context.Service< modelId: providerModel.modelId, name: providerModel.name, configuration: providerModel.configuration, + pricingHistory: providerModel.pricingHistory, }) .from(providerModel) .where( @@ -313,13 +342,21 @@ export class ModelProviders extends Context.Service< ...row, apiKeyHint: apiKey?.slice(-4) ?? null, credentialsReadable: apiKey !== null, - models: models.map((model) => ({ + models: models.map(({ pricingHistory, ...model }) => ({ ...model, + limitOverrides: model.configuration, + priceVersionId: activeModelPrice({ history: pricingHistory, timestamp })?.id ?? null, + scheduledPrices: pricingHistory.filter( + (version): version is ModelScheduledPrice => + version.effectiveFrom > timestamp.toISOString() && version.policy.kind === "manual", + ), configuration: effectiveModelConfiguration({ catalog, providerType: row.providerType, modelId: model.modelId, configured: model.configuration, + history: pricingHistory, + timestamp, }), })), } @@ -378,6 +415,7 @@ export class ModelProviders extends Context.Service< .select({ modelId: providerModel.modelId, configuration: providerModel.configuration, + pricingHistory: providerModel.pricingHistory, }) .from(providerModel) .where( @@ -389,24 +427,6 @@ export class ModelProviders extends Context.Service< .pipe(Effect.orDie) : [] const catalog = yield* readPricing - const configuredModels = input.models.map((model) => ({ - ...model, - configuration: effectiveModelConfiguration({ - catalog, - providerType: input.providerType, - modelId: model.modelId, - configured: - model.configuration?.pricingSource.kind === "manual" - ? model.configuration - : savedModels.find( - (saved) => - saved.modelId === model.modelId && - saved.configuration?.pricingSource.kind === "catalog", - )?.configuration, - }), - })) - if (configuredModels.some((model) => model.configuration === null)) - return yield* new ModelConfigurationMissing() return yield* db .transaction((transaction) => Effect.gen(function* () { @@ -457,6 +477,45 @@ export class ModelProviders extends Context.Service< ), ) } + // A scheduled price may have activated while acquiring the provider lock. + const timestamp = new Date() + const configuredModels: Array<{ + modelId: string + name: string + configuration: ModelLimits | null + pricingHistory: readonly ModelPriceVersion[] + }> = [] + for (const model of input.models) { + const saved = savedModels.find((candidate) => candidate.modelId === model.modelId) + const pricingHistory = updateModelPriceHistory({ + history: saved?.pricingHistory ?? [], + change: model.pricing, + timestamp, + }) + if (!pricingHistory) return yield* new ModelProviderChanged() + const configuration = model.configuration ?? null + const resolved = effectiveModelConfiguration({ + catalog, + providerType: input.providerType, + modelId: model.modelId, + configured: configuration, + history: pricingHistory, + timestamp, + }) + if ( + !resolved && + (!saved || + activeModelPrice({ history: pricingHistory, timestamp })?.policy.kind === + "manual") + ) + return yield* new ModelConfigurationMissing() + configuredModels.push({ + modelId: model.modelId, + name: model.name, + configuration, + pricingHistory, + }) + } const modelIds = input.models.map((model) => model.modelId) yield* transaction .delete(providerModel) @@ -486,6 +545,7 @@ export class ModelProviders extends Context.Service< set: { name: sql`excluded.name`, configuration: sql`excluded.configuration`, + pricingHistory: sql`excluded.pricing_history`, updatedAt: sql`now()`, }, }) @@ -576,7 +636,7 @@ export class ModelProviders extends Context.Service< api: modelProvider.api, baseUrl: modelProvider.baseUrl, modelId: providerModel.modelId, - configured: providerModel.configuration, + pricingModel: { organizationId: providerModel.organizationId, id: providerModel.id }, storedCredentials: sql`${modelProvider.credentials}::text`, }) .from(agentModel) @@ -675,17 +735,34 @@ export class ModelProviders extends Context.Service< if (current?.lockVersion !== input.lockVersion) return yield* new ModelProviderChanged() }) - const pricing = Effect.fn("ModelProviders.pricing")(function* ( - model: Pick, - ) { - if (model.configured?.pricingSource.kind === "manual") return model.configured - const { providerType, modelId, configured } = model - return effectiveModelConfiguration({ - catalog: yield* readPricing, - providerType, - modelId, - configured, - }) + const pricing = Effect.fn("ModelProviders.pricing")(function* (model: ModelPricingInput) { + const [stored] = model.pricingModel + ? yield* db + .select({ + history: providerModel.pricingHistory, + }) + .from(providerModel) + .where( + and( + eq(providerModel.organizationId, model.pricingModel.organizationId), + eq(providerModel.id, model.pricingModel.id), + ), + ) + .limit(1) + .pipe(Effect.orDie) + : [] + if (model.pricingModel && !stored) return null + const timestamp = model.timestamp ?? new Date() + const active = stored && activeModelPrice({ history: stored.history, timestamp }) + if (model.pricingModel && !active) return null + if (active?.policy.kind === "manual") + return { + currency: "USD" as const, + pricingSource: { kind: "manual" as const }, + prices: active.policy.prices, + contextTiers: [], + } + return catalogModelConfiguration({ catalog: yield* readPricing, ...model, timestamp }) }) return ModelProviders.of({ diff --git a/platform/src/lib/model-providers/pricing-catalog-schemas.ts b/platform/src/lib/model-providers/pricing-catalog-schemas.ts index 27840d28..f6c7a2a1 100644 --- a/platform/src/lib/model-providers/pricing-catalog-schemas.ts +++ b/platform/src/lib/model-providers/pricing-catalog-schemas.ts @@ -20,6 +20,17 @@ const catalogMatchSchema: Schema.Codec = Schema.suspend(() => ]), ) const catalogRateSchema = Schema.Number.check(Schema.isFinite(), Schema.isGreaterThanOrEqualTo(0)) +const catalogDateSchema = Schema.String.check( + Schema.isPattern(/^\d{4}-\d{2}-\d{2}$/), + Schema.makeFilter((value) => { + const date = new Date(`${value}T00:00:00Z`) + return ( + !Number.isNaN(date.getTime()) && + date.getUTCFullYear() > 0 && + date.toISOString().startsWith(value) + ) + }), +) const catalogPricesSchema = Schema.Record( Schema.String, Schema.Union([ @@ -53,7 +64,7 @@ const catalogProviderSchema = Schema.Struct({ type: Schema.Literal("start_date").pipe( Schema.withDecodingDefaultKey(Effect.succeed("start_date")), ), - start_date: Schema.String, + start_date: catalogDateSchema, }), Schema.Struct({ type: Schema.Literal("time_of_date").pipe( diff --git a/platform/src/lib/model-providers/pricing-catalog.server.test.ts b/platform/src/lib/model-providers/pricing-catalog.server.test.ts index 7a85a88a..a55c7862 100644 --- a/platform/src/lib/model-providers/pricing-catalog.server.test.ts +++ b/platform/src/lib/model-providers/pricing-catalog.server.test.ts @@ -1,19 +1,11 @@ import { describe, expect, test } from "vitest" +import { Schema } from "effect" import { catalogModelConfiguration, effectiveModelConfiguration } from "./pricing-catalog.server.ts" -import type { ModelPriceCatalog } from "./pricing-catalog-schemas.ts" -import { modelInputAllowance, type ModelConfiguration } from "./schemas.ts" +import { ModelPriceCatalogSchema, type ModelPriceCatalog } from "./pricing-catalog-schemas.ts" +import { modelInputAllowance } from "./schemas.ts" -const modelTestConfiguration: ModelConfiguration = { - currency: "USD", - pricingSource: { kind: "manual" }, - prices: { inputPerMillion: "2", outputPerMillion: "8" }, - contextTiers: [], - maxInputTokens: 128_000, - contextWindowTokens: null, - maxOutputTokens: 8192, - outputCap: 4096, -} +const catalogTestManualPrices = { inputPerMillion: "5", outputPerMillion: "15" } const testPriceCatalog: ModelPriceCatalog = { providers: [ @@ -44,6 +36,49 @@ const testPriceCatalog: ModelPriceCatalog = { } describe("model pricing catalog", () => { + test("selects dated rates at the request start, including the exact boundary", () => { + const catalog: ModelPriceCatalog = { + providers: [ + { + ...testPriceCatalog.providers[0]!, + models: [ + { + ...testPriceCatalog.providers[0]!.models[0]!, + prices: [ + { prices: { input_mtok: 2, output_mtok: 8 } }, + { + constraint: { type: "start_date", start_date: "2026-10-22" }, + prices: { input_mtok: 3, output_mtok: 12 }, + }, + ], + }, + ], + }, + ], + } + const lookup = { catalog, providerType: "openai" as const, modelId: "test-model" } + for (const invalid of ["2026-02-30", "2026-10-22T12:00:00Z"]) { + expect( + Schema.is(ModelPriceCatalogSchema)( + JSON.parse(JSON.stringify(catalog).replace("2026-10-22", invalid)), + ), + ).toBe(false) + } + expect( + catalogModelConfiguration({ + ...lookup, + timestamp: new Date("2026-10-21T23:59:59.999Z"), + })?.prices, + ).toEqual({ inputPerMillion: "2", outputPerMillion: "8" }) + expect( + effectiveModelConfiguration({ + ...lookup, + configured: null, + timestamp: new Date("2026-10-22T00:00:00Z"), + })?.prices, + ).toEqual({ inputPerMillion: "3", outputPerMillion: "12" }) + }) + test("maps supported prices and preserves manual overrides during catalog updates", () => { const lookup = { catalog: testPriceCatalog, @@ -84,23 +119,32 @@ describe("model pricing catalog", () => { ], } expect(catalogModelConfiguration({ ...lookup, catalog: unsupported })).toBeNull() - expect(effectiveModelConfiguration({ ...lookup, configured: modelTestConfiguration })).toBe( - modelTestConfiguration, - ) - const configured = { - ...modelTestConfiguration, - outputCap: 1024, - pricingSource: { kind: "catalog" as const, modelId: "test-model" }, - } - expect(effectiveModelConfiguration({ ...lookup, configured })).toMatchObject({ - outputCap: 4096, + const limits = { maxInputTokens: 128_000, contextWindowTokens: null, + maxOutputTokens: 8192, + outputCap: 1024, + } + const history = [ + { + id: crypto.randomUUID(), + effectiveFrom: "2020-01-01T00:00:00.000Z", + policy: { kind: "manual" as const, prices: catalogTestManualPrices }, + }, + ] + expect(effectiveModelConfiguration({ ...lookup, configured: limits, history })).toMatchObject({ + ...limits, + prices: catalogTestManualPrices, + pricingSource: { kind: "manual" }, + contextTiers: [], + }) + expect(effectiveModelConfiguration({ ...lookup, configured: limits })).toMatchObject({ + ...limits, prices: { inputPerMillion: "2" }, contextTiers: [{ aboveInputTokens: 20_000 }, { aboveInputTokens: 60_000 }], }) - expect(effectiveModelConfiguration({ ...lookup, modelId: "unknown", configured })).toBe( - configured, - ) + expect( + effectiveModelConfiguration({ ...lookup, modelId: "unknown", configured: limits }), + ).toBeNull() }) }) diff --git a/platform/src/lib/model-providers/pricing-catalog.server.ts b/platform/src/lib/model-providers/pricing-catalog.server.ts index 029662ad..136c8333 100644 --- a/platform/src/lib/model-providers/pricing-catalog.server.ts +++ b/platform/src/lib/model-providers/pricing-catalog.server.ts @@ -10,8 +10,11 @@ import { import { ModelConfigurationSchema, type ModelConfiguration, + type ModelLimits, type ModelProviderType, + type ModelPriceVersion, } from "./schemas.ts" +import { activeModelPrice } from "./pricing-history.server.ts" // Catalog provider IDs. https://github.com/pydantic/genai-prices/tree/main/prices/providers export const modelPriceCatalogProviderIds: Record = { @@ -60,12 +63,16 @@ export function catalogModelConfiguration(input: { catalog: ModelPriceCatalog providerType: ModelProviderType modelId: string + timestamp?: Date | undefined }): ModelConfiguration | null { const provider = input.catalog.providers.find( (item) => item.id === modelPriceCatalogProviderIds[input.providerType], ) if (!provider) return null - const result = calcPrice({}, input.modelId, { provider: provider as Provider }) + const result = calcPrice({}, input.modelId, { + provider: provider as Provider, + ...(input.timestamp ? { timestamp: input.timestamp } : {}), + }) if (!result?.model.context_window) return null const prices = result.model_price // Native provider tools are excluded from this rate card. Chat declares function tools only. @@ -119,9 +126,29 @@ export function effectiveModelConfiguration(input: { catalog: ModelPriceCatalog providerType: ModelProviderType modelId: string - configured: ModelConfiguration | null | undefined -}) { - const configured = input.configured - if (configured?.pricingSource.kind === "manual") return configured - return catalogModelConfiguration(input) ?? configured ?? null + timestamp?: Date | undefined + configured: ModelLimits | null | undefined + history?: readonly ModelPriceVersion[] | undefined +}): ModelConfiguration | null { + const active = + input.history && + activeModelPrice({ history: input.history, timestamp: input.timestamp ?? new Date() }) + if (input.history && !active) return null + if (active?.policy.kind === "manual") { + const limits = input.configured ?? catalogModelConfiguration(input) + return limits + ? { + maxInputTokens: limits.maxInputTokens, + contextWindowTokens: limits.contextWindowTokens, + maxOutputTokens: limits.maxOutputTokens, + outputCap: limits.outputCap, + currency: "USD", + pricingSource: { kind: "manual" }, + prices: active.policy.prices, + contextTiers: [], + } + : null + } + const catalog = catalogModelConfiguration(input) + return catalog ? { ...catalog, ...input.configured } : null } diff --git a/platform/src/lib/model-providers/pricing-history.server.test.ts b/platform/src/lib/model-providers/pricing-history.server.test.ts new file mode 100644 index 00000000..86dce229 --- /dev/null +++ b/platform/src/lib/model-providers/pricing-history.server.test.ts @@ -0,0 +1,44 @@ +import { expect, test } from "vitest" +import { updateModelPriceHistory } from "./pricing-history.server" + +test("only changes future prices and cancels them when restoring catalog pricing", () => { + const history = [ + { + id: crypto.randomUUID(), + effectiveFrom: "2026-10-23T10:00:00.000Z", + policy: { kind: "manual" as const, prices: { inputPerMillion: "2", outputPerMillion: "8" } }, + }, + { + id: crypto.randomUUID(), + effectiveFrom: "2026-10-24T10:00:00.000Z", + policy: { kind: "manual" as const, prices: { inputPerMillion: "3", outputPerMillion: "12" } }, + }, + ] + const timestamp = new Date("2026-10-23T11:00:00.000Z") + const change = { expectedVersionId: history[0]!.id } + for (const invalid of [ + { ...history[1]!, id: history[0]!.id }, + { ...history[1]!, effectiveFrom: timestamp.toISOString() }, + ]) { + expect( + updateModelPriceHistory({ + history, + timestamp, + change: { ...change, scheduledPrices: [invalid] }, + }), + ).toBeNull() + } + expect( + updateModelPriceHistory({ + history, + timestamp, + change: { ...change, policy: { kind: "catalog" } }, + }), + ).toEqual([ + history[0], + expect.objectContaining({ + effectiveFrom: timestamp.toISOString(), + policy: { kind: "catalog" }, + }), + ]) +}) diff --git a/platform/src/lib/model-providers/pricing-history.server.ts b/platform/src/lib/model-providers/pricing-history.server.ts new file mode 100644 index 00000000..7b523850 --- /dev/null +++ b/platform/src/lib/model-providers/pricing-history.server.ts @@ -0,0 +1,45 @@ +import { isDeepStrictEqual } from "node:util" + +import type { ModelPriceVersion, ModelPricingChange } from "./schemas.ts" + +export function activeModelPrice(input: { + history: readonly ModelPriceVersion[] + timestamp: Date +}) { + return input.history.findLast((version) => version.effectiveFrom <= input.timestamp.toISOString()) +} + +/** Apply explicit pricing edits under the provider lock. Effective versions stay immutable. */ +export function updateModelPriceHistory(input: { + history: readonly ModelPriceVersion[] + change: ModelPricingChange | undefined + timestamp: Date +}): readonly ModelPriceVersion[] | null { + if (input.history.length && !input.change) return input.history + const now = input.timestamp.toISOString() + const active = activeModelPrice(input) + const { policy, scheduledPrices } = input.change ?? {} + if (policy && input.change!.expectedVersionId !== (active?.id ?? null)) return null + const past = input.history.filter((version) => version.effectiveFrom <= now) + const future = + policy?.kind === "catalog" + ? [] + : (scheduledPrices ?? input.history.filter((version) => version.effectiveFrom > now)) + if ( + future.some( + (version) => version.effectiveFrom <= now || past.some(({ id }) => id === version.id), + ) + ) + return null + if (!active || (policy && !isDeepStrictEqual(active.policy, policy))) { + if (active?.effectiveFrom === now) return null + past.push({ + id: crypto.randomUUID(), + effectiveFrom: now, + policy: policy ?? { kind: "catalog" }, + }) + } + return [...past, ...future].toSorted((left, right) => + left.effectiveFrom.localeCompare(right.effectiveFrom), + ) +} diff --git a/platform/src/lib/model-providers/schemas.ts b/platform/src/lib/model-providers/schemas.ts index c342d34f..6f15d3af 100644 --- a/platform/src/lib/model-providers/schemas.ts +++ b/platform/src/lib/model-providers/schemas.ts @@ -21,7 +21,7 @@ const ModelTokenPricesSchema = Schema.Struct({ cacheWrite1hPerMillion: Schema.optionalKey(modelPriceSchema), }) -export const ModelConfigurationSchema = Schema.Struct({ +const ModelRateCardSchema = Schema.Struct({ currency: Schema.Literal("USD"), pricingSource: Schema.Union([ Schema.Struct({ kind: Schema.Literal("catalog"), modelId: Schema.String }), @@ -34,26 +34,38 @@ export const ModelConfigurationSchema = Schema.Struct({ prices: ModelTokenPricesSchema, }), ).check(Schema.isMaxLength(32)), +}) + +export type ModelRateCard = typeof ModelRateCardSchema.Type + +export const ModelLimitsSchema = Schema.Struct({ maxInputTokens: Schema.NullOr(modelTokenBoundSchema), contextWindowTokens: Schema.NullOr(modelTokenBoundSchema), maxOutputTokens: modelTokenBoundSchema, outputCap: modelTokenBoundSchema, }).check( + Schema.makeFilter((limits) => { + if (limits.maxInputTokens === null && limits.contextWindowTokens === null) + return { path: ["maxInputTokens"], issue: "Configure an input or context token limit" } + if (limits.outputCap > limits.maxOutputTokens) + return { path: ["outputCap"], issue: "Output cap exceeds the configured output maximum" } + if (limits.contextWindowTokens !== null && limits.outputCap >= limits.contextWindowTokens) + return { path: ["outputCap"], issue: "Leave room for input within the context window" } + return undefined + }), +) +export type ModelLimits = typeof ModelLimitsSchema.Type + +export const ModelConfigurationSchema = ModelLimitsSchema.mapFields( + (fields) => ({ ...fields, ...ModelRateCardSchema.fields }), + { unsafePreserveChecks: true }, +).check( Schema.makeFilter((configuration) => { if (configuration.pricingSource.kind === "manual" && configuration.contextTiers.length > 0) return { path: ["contextTiers"], issue: "Manual overrides use flat prices without context tiers", } - if (configuration.maxInputTokens === null && configuration.contextWindowTokens === null) - return { path: ["maxInputTokens"], issue: "Configure an input or context token limit" } - if (configuration.outputCap > configuration.maxOutputTokens) - return { path: ["outputCap"], issue: "Output cap exceeds the configured output maximum" } - if ( - configuration.contextWindowTokens !== null && - configuration.outputCap >= configuration.contextWindowTokens - ) - return { path: ["outputCap"], issue: "Leave room for input within the context window" } if ( configuration.contextTiers.some( (tier, index, tiers) => @@ -64,9 +76,56 @@ export const ModelConfigurationSchema = Schema.Struct({ return undefined }), ) - export type ModelConfiguration = typeof ModelConfigurationSchema.Type +const modelPriceInstant = Schema.String.check( + Schema.isPattern(/^\d{4}-/), + Schema.makeFilter((value) => { + const date = new Date(value) + return !Number.isNaN(date.getTime()) && date.toISOString() === value + }), +) +const ModelManualPricePolicySchema = Schema.Struct({ + kind: Schema.Literal("manual"), + prices: ModelTokenPricesSchema, +}) +export const ModelPricePolicySchema = Schema.Union([ + Schema.Struct({ kind: Schema.Literal("catalog") }), + ModelManualPricePolicySchema, +]) +const ModelPriceVersionSchema = Schema.Struct({ + id: Schema.String.check(Schema.isUUID()), + effectiveFrom: modelPriceInstant, + policy: ModelPricePolicySchema, +}) +export type ModelPriceVersion = typeof ModelPriceVersionSchema.Type +export const ModelPriceHistorySchema = Schema.Array(ModelPriceVersionSchema).check( + Schema.makeFilter( + (versions) => + new Set(versions.map(({ effectiveFrom }) => effectiveFrom)).size === versions.length && + new Set(versions.map(({ id }) => id)).size === versions.length, + ), +) +const ModelScheduledPriceSchema = Schema.Struct({ + ...ModelPriceVersionSchema.fields, + policy: ModelManualPricePolicySchema, +}) +export type ModelScheduledPrice = typeof ModelScheduledPriceSchema.Type +const ModelScheduledPricesSchema = Schema.Array(ModelScheduledPriceSchema).check( + Schema.isMaxLength(32), + Schema.makeFilter( + (versions) => + new Set(versions.map(({ id }) => id)).size === versions.length && + new Set(versions.map(({ effectiveFrom }) => effectiveFrom)).size === versions.length, + ), +) +const ModelPricingChangeSchema = Schema.Struct({ + expectedVersionId: Schema.NullOr(Schema.String.check(Schema.isUUID())), + policy: Schema.optionalKey(ModelPricePolicySchema), + scheduledPrices: Schema.optionalKey(ModelScheduledPricesSchema), +}) +export type ModelPricingChange = typeof ModelPricingChangeSchema.Type + export function modelInputAllowance(configuration: ModelConfiguration) { return Math.min( configuration.maxInputTokens ?? Infinity, @@ -86,6 +145,22 @@ const ModelProviderApiSchema = enumSchema([ "gemini-generate-content", ]) export type ModelProviderApi = typeof ModelProviderApiSchema.Type + +export const ModelRequestSnapshotSchema = Schema.Struct({ + id: Schema.String.check(Schema.isUUID()), + startedAt: modelPriceInstant, + providerId: Schema.String, + providerType: ModelProviderTypeSchema, + protocol: ModelProviderApiSchema, + modelId: Schema.String, + rates: Schema.NullOr(ModelRateCardSchema), + usage: Schema.optionalKey(Schema.JsonObject), + costUsd: Schema.optionalKey( + Schema.NullOr(Schema.String.check(Schema.isPattern(/^-?\d+(?:\.\d+)?$/))), + ), +}) +export type ModelRequestSnapshot = typeof ModelRequestSnapshotSchema.Type + const modelProviderApis: Record = { openai: ["responses", "chat-completions"], anthropic: ["anthropic-messages"], @@ -129,6 +204,8 @@ const ProviderModelFieldsSchema = Schema.Struct({ ), name: DisplayNameSchema, configuration: Schema.optionalKey(Schema.NullOr(ModelConfigurationSchema)), + priceVersionId: Schema.optionalKey(Schema.NullOr(Schema.String.check(Schema.isUUID()))), + scheduledPrices: Schema.optionalKey(ModelScheduledPricesSchema), }) export type ProviderModelFields = typeof ProviderModelFieldsSchema.Type @@ -170,7 +247,6 @@ export const ModelProviderFieldsSchema = Schema.Struct({ }), ), ) -export type ModelProviderFields = typeof ModelProviderFieldsSchema.Type export const ModelProviderCredentialsPayloadSchema = Schema.Struct({ organizationId: UuidV7Schema, @@ -178,3 +254,24 @@ export const ModelProviderCredentialsPayloadSchema = Schema.Struct({ providerType: ModelProviderTypeSchema, apiKey: ModelProviderApiKeySchema, }) + +export const ModelProviderWriteFieldsSchema = ModelProviderFieldsSchema.mapFields( + (fields) => ({ + ...fields, + models: Schema.Array( + Schema.Struct({ + modelId: ProviderModelFieldsSchema.fields.modelId, + name: DisplayNameSchema, + configuration: Schema.optionalKey(Schema.NullOr(ModelLimitsSchema)), + pricing: Schema.optionalKey(ModelPricingChangeSchema), + }), + ).check( + Schema.isMaxLength(100), + Schema.makeFilter( + (models) => new Set(models.map(({ modelId }) => modelId)).size === models.length, + ), + ), + }), + { unsafePreserveChecks: true }, +) +export type ModelProviderWriteFields = typeof ModelProviderWriteFieldsSchema.Type diff --git a/platform/src/lib/model-providers/usage-cost.ts b/platform/src/lib/model-providers/usage-cost.ts index 53243c8a..07d72ae3 100644 --- a/platform/src/lib/model-providers/usage-cost.ts +++ b/platform/src/lib/model-providers/usage-cost.ts @@ -1,8 +1,8 @@ import { BigDecimal, Schema } from "effect" -import type { ModelConfiguration } from "./schemas.ts" +import type { ModelRateCard } from "./schemas.ts" -export const ModelRequestUsageSchema = Schema.Struct({ +const ModelRequestUsageSchema = Schema.Struct({ promptTokens: Schema.Number, completionTokens: Schema.Number, promptTokensDetails: Schema.optional( @@ -21,11 +21,24 @@ export type ModelCallUsage = typeof ModelRequestUsageSchema.Type & { const perMillion = BigDecimal.fromNumberUnsafe(1_000_000) +export function modelRequestCost(input: { + usage: ModelCallUsage & { readonly cost?: number | undefined } + configuration: ModelRateCard | null | undefined +}): string | null { + const reported = input.usage.providerUsageDetails?.costUsd + if (typeof reported === "string" && /^\d+(?:\.\d+)?$/.test(reported)) return reported + return typeof input.usage.cost === "number" && Number.isFinite(input.usage.cost) + ? BigDecimal.format(BigDecimal.fromNumberUnsafe(input.usage.cost)) + : input.configuration + ? modelUsageCost(input.configuration, input.usage) + : null +} + /** * The US dollar token cost at a configuration's rates, as a decimal string. Context tiers price each * underlying request by its prompt size, and missing cache rates fall back to the input rate. */ -export function modelUsageCost(configuration: ModelConfiguration, usage: ModelCallUsage): string { +export function modelUsageCost(configuration: ModelRateCard, usage: ModelCallUsage): string { const requests = usage.providerUsageDetails?.requestUsage if (Schema.is(Schema.Array(ModelRequestUsageSchema))(requests)) { return BigDecimal.format( diff --git a/platform/src/lib/tenants/tenant-users.ts b/platform/src/lib/tenants/tenant-users.ts index 932e8427..aa6b5e62 100644 --- a/platform/src/lib/tenants/tenant-users.ts +++ b/platform/src/lib/tenants/tenant-users.ts @@ -1,18 +1,12 @@ import { and, asc, desc, eq, gt, gte, ilike, inArray, lt, ne, or, sql } from "drizzle-orm" -import { BigDecimal, Context, DateTime, Effect, Layer, Schema } from "effect" +import { BigDecimal, Context, DateTime, Effect, Layer } from "effect" import { Database } from "@/db/database" import { type DatabasePage, type DatabasePageOptions, databasePage } from "@/db/lib/pagination" import { mapDatabaseErrors } from "@/db/lib/sqlstate" import { chatMessage } from "@/db/schema/chat" -import { modelProvider, providerModel, tenant, tenantUser } from "@/db/schema/organizations" +import { tenant, tenantUser } from "@/db/schema/organizations" import type { ChatPrincipal } from "@/lib/chat/types" -import { Config } from "@/lib/config/config" -import { - effectiveModelConfiguration, - readModelPriceCatalog, -} from "@/lib/model-providers/pricing-catalog.server" -import { ModelRequestUsageSchema, modelUsageCost } from "@/lib/model-providers/usage-cost" import { TenantExternalIdTaken, TenantNotFound, TenantUserNotFound } from "./errors.ts" import type { TenantUserPatchSchema, @@ -117,7 +111,6 @@ export class TenantUsers extends Context.Service< Effect.gen(function* () { const db = yield* Database const tenants = yield* Tenants - const config = yield* Config // A Tenant-scoped credential reaches only its own Tenant, without another query. const requireTenant = (scope: TenantScope, tenantId: string) => @@ -172,130 +165,51 @@ export class TenantUsers extends Context.Service< when jsonb_typeof(${savedUsage}->'cost') = 'number' then (${savedUsage}->>'cost')::numeric end` - const range = and( - eq(chatMessage.organizationId, scope.organizationId), - eq(chatMessage.tenantId, tenantId), - eq(chatMessage.role, "assistant"), - ne(chatMessage.state, "draft"), - gte(chatMessage.createdAt, DateTime.toDateUtc(DateTime.makeUnsafe(start))), - lt(chatMessage.createdAt, DateTime.toDateUtc(DateTime.makeUnsafe(end))), - ) + const startAt = DateTime.toDateUtc(DateTime.makeUnsafe(start)) + const endAt = DateTime.toDateUtc(DateTime.makeUnsafe(end)) + const perRequest = sql`${chatMessage.requestAccounting}->>'version' = '2'` + const cost = sql`case when ${perRequest} then + (select coalesce(sum((attempt->>'costUsd')::numeric), 0) + from jsonb_array_elements(${chatMessage.requestAccounting}->'requests') attempt + where (attempt->>'startedAt')::timestamptz >= ${startAt} + and (attempt->>'startedAt')::timestamptz < ${endAt} + and attempt->>'costUsd' ~ '^-?[0-9]+([.][0-9]+)?$') + else ${savedCost} end` const totals = yield* db - .select({ - tenantUserId: initiator, - cost: sql`coalesce(sum(${savedCost}), 0)::text`, - needsPricing: sql`bool_or(${savedCost} is null)`, - }) + .select({ tenantUserId: initiator, cost: sql`coalesce(sum(${cost}), 0)::text` }) .from(chatMessage) .where( and( - range, + eq(chatMessage.organizationId, scope.organizationId), + eq(chatMessage.tenantId, tenantId), + eq(chatMessage.role, "assistant"), + ne(chatMessage.state, "draft"), inArray( initiator, users.map(({ id }) => id), ), + or( + and( + perRequest, + gte(chatMessage.lastRequestAt, startAt), + lt(chatMessage.firstRequestAt, endAt), + ), + and( + sql`(${perRequest}) is not true`, + gte(chatMessage.createdAt, startAt), + lt(chatMessage.createdAt, endAt), + ), + ), ), ) .groupBy(initiator) - const costs = new Map( - users.map(({ id }) => [id, BigDecimal.fromNumberUnsafe(0)]), - ) - for (const total of totals) - costs.set(total.tenantUserId, BigDecimal.fromStringUnsafe(total.cost)) - const unpriced = totals - .filter((total) => total.needsPricing) - .map((total) => total.tenantUserId) - const catalog = - unpriced.length > 0 - ? yield* readModelPriceCatalog.pipe(Effect.provideService(Config, config)) - : undefined - const pricingByModel = new Map>() - let cursor: - | { tenantUserId: string; createdAt: string; threadId: string; id: string } - | undefined - while (catalog) { - const replies = yield* db - .select({ - id: chatMessage.id, - threadId: chatMessage.threadId, - createdAt: sql`${chatMessage.createdAt}::text`, - tenantUserId: initiator, - usage: sql`${chatMessage.metadata}->'provenance'->'usage'`, - providerId: modelProvider.id, - providerType: modelProvider.providerType, - modelId: providerModel.modelId, - configuration: providerModel.configuration, - }) - .from(chatMessage) - .leftJoin( - modelProvider, - and( - eq(modelProvider.organizationId, chatMessage.organizationId), - sql`${modelProvider.id}::text = ${chatMessage.metadata}->'provenance'->>'providerId'`, - eq( - modelProvider.providerType, - sql`${chatMessage.metadata}->'provenance'->>'providerType'`, - ), - ), - ) - .leftJoin( - providerModel, - and( - eq(providerModel.organizationId, chatMessage.organizationId), - eq(providerModel.modelProviderId, modelProvider.id), - eq(providerModel.modelId, sql`${chatMessage.metadata}->'provenance'->>'modelId'`), - ), - ) - .where( - and( - range, - inArray(initiator, unpriced), - sql`${savedCost} is null`, - cursor - ? sql`(${initiator}, ${chatMessage.createdAt}, ${chatMessage.threadId}, ${chatMessage.id}) > (${cursor.tenantUserId}, ${cursor.createdAt}::timestamptz, ${cursor.threadId}::uuid, ${cursor.id}::uuid)` - : undefined, - ), - ) - .orderBy( - asc(initiator), - asc(chatMessage.createdAt), - asc(chatMessage.threadId), - asc(chatMessage.id), - ) - .limit(5000) - for (const { - tenantUserId, - usage, - providerId, - providerType, - modelId, - configuration, - } of replies) { - if (!providerType || !modelId || !Schema.is(ModelRequestUsageSchema)(usage)) continue - const key = `${providerId}:${modelId}` - let pricing = pricingByModel.get(key) - if (pricing === undefined) { - pricing = effectiveModelConfiguration({ - catalog, - providerType, - modelId, - configured: configuration, - }) - pricingByModel.set(key, pricing) - } - if (!pricing) continue - const cost = BigDecimal.fromStringUnsafe(modelUsageCost(pricing, usage)) - costs.set(tenantUserId, BigDecimal.sum(costs.get(tenantUserId)!, cost)) - } - if (replies.length < 5000) break - cursor = replies.at(-1)! - } + const costs = new Map(totals.map(({ tenantUserId, cost }) => [tenantUserId, cost])) return { start, end, - items: Array.from(costs, ([tenantUserId, cost]) => ({ - tenantUserId, - costUsd: BigDecimal.format(cost), + items: users.map(({ id }) => ({ + tenantUserId: id, + costUsd: BigDecimal.format(BigDecimal.fromStringUnsafe(costs.get(id) ?? "0")), })), } }, mapDatabaseErrors()) @@ -409,6 +323,6 @@ export class TenantUsers extends Context.Service< ) static readonly layer = TenantUsers.layerNoDeps.pipe( - Layer.provide([Tenants.layer, Database.layer, Config.layer]), + Layer.provide([Tenants.layer, Database.layer]), ) } diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-fields.tsx b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-fields.tsx new file mode 100644 index 00000000..542ffbf7 --- /dev/null +++ b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-fields.tsx @@ -0,0 +1,59 @@ +import type { ModelConfiguration } from "@/lib/model-providers/schemas" +import { ProviderField } from "../../-components/provider-field" + +const modelPriceFields = [ + ["inputPerMillion", "Input price"], + ["outputPerMillion", "Output price"], + ["cacheReadPerMillion", "Cache read price"], + ["cacheWritePerMillion", "Cache write price"], + ["cacheWrite1hPerMillion", "One-hour cache write price"], +] as const + +export function ModelPriceFields({ + id, + prices, + disabled, + errors, + onChange, +}: { + id: string + prices: ModelConfiguration["prices"] + disabled: boolean + errors: (field: string) => readonly { message: string }[] + onChange: (prices: ModelConfiguration["prices"]) => void +}) { + const fields = modelPriceFields.map(([key, label]) => ( + + {label} + for {id} + + } + inputProps={{ inputMode: "decimal" }} + value={prices[key] ?? ""} + disabled={disabled} + errors={errors(key)} + placeholder={ + key === "inputPerMillion" || key === "outputPerMillion" ? "Required" : "Optional" + } + onChange={(value) => { + const next = { ...prices } + if (!value && key !== "inputPerMillion" && key !== "outputPerMillion") delete next[key] + else next[key] = value + onChange(next) + }} + /> + )) + return ( + <> +
{fields.slice(0, 2)}
+
+ Cache pricing +
{fields.slice(2)}
+
+ + ) +} diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-schedule.tsx b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-schedule.tsx new file mode 100644 index 00000000..156c5c95 --- /dev/null +++ b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-price-schedule.tsx @@ -0,0 +1,145 @@ +import { XIcon } from "@phosphor-icons/react" +import { useState } from "react" +import { useIsHydrated } from "@/components/auth/use-is-hydrated" +import { Button } from "@/components/ui/button" +import { FieldDescription, FieldError } from "@/components/ui/field" +import { LocalDateTime } from "@/components/local-date-time" +import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip" +import type { ProviderModelFields } from "@/lib/model-providers/schemas" +import { ProviderField } from "../../-components/provider-field" +import { ModelPriceFields } from "./model-price-fields" + +export function ModelPriceSchedule({ + model, + disabled, + errors, + onChange, +}: { + model: ProviderModelFields + disabled: boolean + errors: readonly { message: string }[] + onChange: (prices: NonNullable) => void +}) { + const schedule = model.scheduledPrices ?? [] + const hydrated = useIsHydrated() + const [expanded, setExpanded] = useState(null) + const [minimum] = useState(() => + new Date(Date.now() - new Date().getTimezoneOffset() * 60_000 + 60_000) + .toISOString() + .slice(0, 16), + ) + return ( +
+

Price schedule

+ + Current price changes take effect when saved. Future prices use your time zone. Active + versions cannot be edited or deleted. + + {schedule.map((version) => { + const date = new Date(version.effectiveFrom) + const value = + !hydrated || Number.isNaN(date.getTime()) + ? "" + : new Date(date.getTime() - date.getTimezoneOffset() * 60_000) + .toISOString() + .slice(0, 16) + return ( +
{ + const open = event.currentTarget.open + setExpanded((current) => + open ? version.id : current === version.id ? null : current, + ) + }} + className="space-y-3 rounded-md border p-3" + > + + {version.effectiveFrom ? ( + <> + Scheduled ยท + + ) : ( + "New scheduled price" + )} + +
+
+ + onChange( + schedule.map((entry) => + entry.id === version.id + ? { ...entry, effectiveFrom: next ? new Date(next).toISOString() : "" } + : entry, + ), + ) + } + /> +
+ + onChange(schedule.filter((entry) => entry.id !== version.id))} + /> + } + > + + + Cancel scheduled price + +
+ []} + onChange={(prices) => + onChange( + schedule.map((entry) => + entry.id === version.id + ? { ...entry, policy: { kind: "manual", prices } } + : entry, + ), + ) + } + /> +
+ ) + })} + + +
+ ) +} diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-components/model-provider-form.tsx b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-provider-form.tsx index e1396b6d..8934f5f8 100644 --- a/platform/src/routes/_authenticated/$orgSlug/models/-components/model-provider-form.tsx +++ b/platform/src/routes/_authenticated/$orgSlug/models/-components/model-provider-form.tsx @@ -33,18 +33,24 @@ import { type ModelProviderApi, type ModelProviderType, type ProviderModelFields, + ModelPricePolicySchema, + ModelPriceHistorySchema, + ModelLimitsSchema, } from "@/lib/model-providers/schemas" import { parseServerFnError, type ServerFnError } from "@/lib/runtime/server-fn-error" import { strictParseOptions } from "@/lib/schemas" import { saveModelProvider } from "../-functions/save-model-provider" import { modelProviderApiItems, modelProviderDescriptors } from "../-lib/constants" -import { SaveModelProviderInputSchema } from "../-lib/schemas" +import { ModelProviderFormInputSchema } from "../-lib/schemas" import { ProviderField } from "../../-components/provider-field" import { ProviderModelPicker } from "./provider-model-picker" import { ModelProviderTest } from "./model-provider-test" import { ProviderModelConfigurationFields } from "./provider-model-configuration-fields" const formatModelProviderIssues = SchemaIssue.makeFormatterStandardSchemaV1() +const equalModelLimits = Schema.toEquivalence(ModelLimitsSchema) +const equalModelPricePolicy = Schema.toEquivalence(ModelPricePolicySchema) +const equalModelPriceHistory = Schema.toEquivalence(ModelPriceHistorySchema) const equalModelProviderFields = Schema.toEquivalence(ModelProviderFieldsSchema) const modelProviderTypeItems = Object.entries(modelProviderDescriptors).map( ([value, { label }]) => ({ label, value: value as ModelProviderType }), @@ -72,11 +78,15 @@ export function ModelProviderForm({ const [baseUrl, setBaseUrl] = useState(existing?.baseUrl ?? "https://api.openai.com/v1") const [apiKey, setApiKey] = useState("") const [models, setModels] = useState( - existing?.models.map(({ modelId, name: modelName, configuration }) => ({ - modelId, - name: modelName, - configuration, - })) ?? [], + existing?.models.map( + ({ modelId, name: modelName, configuration, priceVersionId, scheduledPrices }) => ({ + modelId, + name: modelName, + configuration, + priceVersionId, + scheduledPrices, + }), + ) ?? [], ) const [overridesOpen, setOverridesOpen] = useState( existing !== null && @@ -92,7 +102,7 @@ export function ModelProviderForm({ const [saveError, setSaveError] = useState(null) const conflicted = saveError?.tag === "ModelProviderChanged" const input = Schema.decodeUnknownResult( - SaveModelProviderInputSchema, + ModelProviderFormInputSchema, strictParseOptions, )({ organizationSlug, @@ -145,7 +155,57 @@ export function ModelProviderForm({ setServerFieldError(null) setSaveError(null) try { - const modelProviderId = await saveModelProvider({ data: input.success }) + const modelProviderId = await saveModelProvider({ + data: { + ...input.success, + models: input.success.models.map((model) => { + const previous = existing?.models.find(({ modelId }) => modelId === model.modelId) + const policy = + model.configuration?.pricingSource.kind === "manual" + ? { kind: "manual" as const, prices: model.configuration.prices } + : { kind: "catalog" as const } + const before = + previous?.configuration?.pricingSource.kind === "manual" + ? { kind: "manual" as const, prices: previous.configuration.prices } + : { kind: "catalog" as const } + const changed = !previous || !equalModelPricePolicy(policy, before) + const scheduled = !equalModelPriceHistory( + model.scheduledPrices ?? [], + previous?.scheduledPrices ?? [], + ) + const limits = model.configuration + const beforeLimits = + previous?.configuration ?? + catalog[providerType].find(({ modelId }) => modelId === model.modelId)?.configuration + const customLimits = + limits && (!beforeLimits || !equalModelLimits(limits, beforeLimits)) + return { + modelId: model.modelId, + name: model.name, + configuration: + limits?.pricingSource.kind === "manual" + ? customLimits + ? { + maxInputTokens: limits.maxInputTokens, + contextWindowTokens: limits.contextWindowTokens, + maxOutputTokens: limits.maxOutputTokens, + outputCap: limits.outputCap, + } + : (previous?.limitOverrides ?? null) + : null, + ...(changed || scheduled + ? { + pricing: { + expectedVersionId: previous?.priceVersionId ?? null, + ...(changed ? { policy } : {}), + ...(scheduled ? { scheduledPrices: model.scheduledPrices ?? [] } : {}), + }, + } + : {}), + } + }), + }, + }) toast.add({ title: "Provider saved", type: "success" }) if (existing) { await router.invalidate() @@ -329,6 +389,23 @@ export function ModelProviderForm({ ?.configuration ?? null } disabled={disabled} + scheduleErrors={ + submitted + ? issues.filter( + (issue) => + issue.path?.[0] === "models" && + issue.path?.[1] === index && + issue.path?.[2] === "scheduledPrices", + ) + : [] + } + onScheduleChange={(scheduledPrices) => + setModels((current) => + current.map((item) => + item.modelId === model.modelId ? { ...item, scheduledPrices } : item, + ), + ) + } errors={(field) => submitted ? issues.filter( @@ -343,7 +420,15 @@ export function ModelProviderForm({ onChange={(configuration) => setModels((current) => current.map((item) => - item.modelId === model.modelId ? { ...item, configuration } : item, + item.modelId === model.modelId + ? { + ...item, + configuration, + ...(configuration?.pricingSource.kind !== "manual" + ? { scheduledPrices: [] } + : {}), + } + : item, ), ) } @@ -364,7 +449,7 @@ export function ModelProviderForm({ type="button" variant="outline" className="mt-2 w-fit" - onClick={() => void router.invalidate()} + onClick={() => window.location.reload()} > Reload latest @@ -395,11 +480,21 @@ export function ModelProviderForm({ setBaseUrl(existing.baseUrl) setApiKey("") setModels( - existing.models.map(({ modelId, name: modelName, configuration }) => ({ - modelId, - name: modelName, - configuration, - })), + existing.models.map( + ({ + modelId, + name: modelName, + configuration, + priceVersionId, + scheduledPrices, + }) => ({ + modelId, + name: modelName, + configuration, + priceVersionId, + scheduledPrices, + }), + ), ) setSubmitted(false) setServerFieldError(null) diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-components/provider-model-configuration-fields.tsx b/platform/src/routes/_authenticated/$orgSlug/models/-components/provider-model-configuration-fields.tsx index 71d10b91..b65e5818 100644 --- a/platform/src/routes/_authenticated/$orgSlug/models/-components/provider-model-configuration-fields.tsx +++ b/platform/src/routes/_authenticated/$orgSlug/models/-components/provider-model-configuration-fields.tsx @@ -8,14 +8,9 @@ import { type ProviderModelFields, } from "@/lib/model-providers/schemas" import { ProviderField } from "../../-components/provider-field" +import { ModelPriceFields } from "./model-price-fields" +import { ModelPriceSchedule } from "./model-price-schedule" -const modelPricingFields = [ - ["inputPerMillion", "Input price"], - ["outputPerMillion", "Output price"], - ["cacheReadPerMillion", "Cache read price"], - ["cacheWritePerMillion", "Cache write price"], - ["cacheWrite1hPerMillion", "One-hour cache write price"], -] as const const modelBoundFields = [ ["maxInputTokens", "Input maximum"], ["contextWindowTokens", "Context window"], @@ -29,12 +24,16 @@ export function ProviderModelConfigurationFields({ catalogConfiguration, errors, onChange, + onScheduleChange, + scheduleErrors, }: { model: ProviderModelFields disabled: boolean catalogConfiguration: ModelConfiguration | null errors: (field: string) => readonly { message: string }[] onChange: (configuration: ModelConfiguration | null) => void + onScheduleChange: (scheduledPrices: NonNullable) => void + scheduleErrors: readonly { message: string }[] }) { const overridden = model.configuration?.pricingSource.kind === "manual" const configuration: ModelConfiguration = model.configuration ?? @@ -50,31 +49,6 @@ export function ProviderModelConfigurationFields({ } const hasCatalogDefaults = catalogConfiguration !== null const overrideId = `model-${model.modelId}-override` - const priceFields = modelPricingFields.map(([key, label]) => ( - - {label} - for {model.modelId} - - } - inputProps={{ inputMode: "decimal" }} - value={configuration.prices[key] ?? ""} - disabled={disabled} - errors={errors(key)} - placeholder={ - key === "inputPerMillion" || key === "outputPerMillion" ? "Required" : "Optional" - } - onChange={(value) => { - const prices = { ...configuration.prices } - if (!value && key !== "inputPerMillion" && key !== "outputPerMillion") delete prices[key] - else prices[key] = value - onChange({ ...configuration, prices }) - }} - /> - )) return (
{overridden ? ( <> -
{priceFields.slice(0, 2)}
+ onChange({ ...configuration, prices })} + />

Token limits

{modelBoundFields.map(([key, label]) => ( @@ -148,10 +128,12 @@ export function ProviderModelConfigurationFields({ maximum. Verify the default 4,096-token output maximum with your provider before increasing it. -
- Cache pricing -
{priceFields.slice(2)}
-
+ ) : ( hasCatalogDefaults && ( diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-functions/get-model-provider-page-data.ts b/platform/src/routes/_authenticated/$orgSlug/models/-functions/get-model-provider-page-data.ts index 60cc9fc9..f520777b 100644 --- a/platform/src/routes/_authenticated/$orgSlug/models/-functions/get-model-provider-page-data.ts +++ b/platform/src/routes/_authenticated/$orgSlug/models/-functions/get-model-provider-page-data.ts @@ -7,7 +7,7 @@ import { Effect } from "effect" import { ModelProviders } from "@/lib/model-providers/model-providers.server" import { - effectiveModelConfiguration, + catalogModelConfiguration, readModelPriceCatalog, } from "@/lib/model-providers/pricing-catalog.server" import type { ModelProviderType } from "@/lib/model-providers/schemas" @@ -89,12 +89,10 @@ export const getModelProviderPageData = createServerFn({ method: "GET" }) ].map((model) => ({ modelId: model.modelId, name: model.name, - configuration: effectiveModelConfiguration({ + configuration: catalogModelConfiguration({ catalog: pricing, providerType, modelId: model.modelId, - configured: - model.configuration?.pricingSource.kind === "catalog" ? model.configuration : null, }), })) const catalog = { diff --git a/platform/src/routes/_authenticated/$orgSlug/models/-lib/schemas.ts b/platform/src/routes/_authenticated/$orgSlug/models/-lib/schemas.ts index 900b2d34..59080016 100644 --- a/platform/src/routes/_authenticated/$orgSlug/models/-lib/schemas.ts +++ b/platform/src/routes/_authenticated/$orgSlug/models/-lib/schemas.ts @@ -1,10 +1,23 @@ import { Schema } from "effect" -import { ModelProviderFieldsSchema } from "@/lib/model-providers/schemas" +import { + ModelProviderFieldsSchema, + ModelProviderWriteFieldsSchema, +} from "@/lib/model-providers/schemas" import { SlugSchema } from "@/lib/organizations/slug" import { LockVersionSchema, UuidV7Schema } from "@/lib/schemas" -export const SaveModelProviderInputSchema = ModelProviderFieldsSchema.mapFields( +export const ModelProviderFormInputSchema = ModelProviderFieldsSchema.mapFields( + (fields) => ({ + ...fields, + organizationSlug: SlugSchema, + id: Schema.NullOr(UuidV7Schema), + lockVersion: Schema.NullOr(LockVersionSchema), + }), + { unsafePreserveChecks: true }, +) + +export const SaveModelProviderInputSchema = ModelProviderWriteFieldsSchema.mapFields( (fields) => ({ ...fields, organizationSlug: SlugSchema, diff --git a/platform/src/routes/api/v1/-lib/transport.test.ts b/platform/src/routes/api/v1/-lib/transport.test.ts index 8134f18e..a7945041 100644 --- a/platform/src/routes/api/v1/-lib/transport.test.ts +++ b/platform/src/routes/api/v1/-lib/transport.test.ts @@ -55,7 +55,6 @@ const restTestState = vi.hoisted(() => ({ writes: [] as unknown[], order: [] as SQL[], limits: [] as number[], - usageReplies: [] as unknown[], usageTotals: [] as unknown[], usagePredicates: [] as SQL[], keyRows: [] as { id: string; name?: string; slug?: string }[], @@ -165,8 +164,6 @@ function restTestDatabase(): EffectDatabase { return Effect.succeed(restTestState.usageTotals) }, limit(limit: number) { - if (this.table === chatMessage) - return Effect.succeed(restTestState.usageReplies.splice(0, limit)) if (this.table === organization) return Effect.succeed(restTestState.keyRows) restTestState.limits.push(limit) return result() @@ -342,6 +339,9 @@ const restSavedMessage: MessageRecord = { turnMessageId: null, turnState: "completed", metadata: { version: 1 }, + requestAccounting: null, + firstRequestAt: null, + lastRequestAt: null, payload: { version: 1, parts: [{ id: "part", type: "text", content: "Saved history" }] }, sourceAssistantMessageId: null, sourceToolPartId: null, @@ -428,7 +428,6 @@ beforeEach(() => { writes: [], order: [], limits: [], - usageReplies: [], usageTotals: [], usagePredicates: [], logs: [], @@ -1460,47 +1459,10 @@ describe("REST API through the Effect Fetch handler", () => { expect(restTestState.usagePredicates).toHaveLength(0) }) - test("usage batches legacy pricing and treats missing costs as zero", async () => { - const pricedReply = { - tenantUserId: restUserId, - id: restUserId, - threadId: restTenantId, - createdAt: "2024-02-29 07:04:56.123456+00", - usage: { promptTokens: 1000, completionTokens: 0 }, - providerId: restOrgId, - providerType: "openai", - modelId: "custom", - configuration: { - currency: "USD", - pricingSource: { kind: "manual" }, - prices: { inputPerMillion: "3", outputPerMillion: "15" }, - contextTiers: [], - maxInputTokens: 100_000, - contextWindowTokens: null, - maxOutputTokens: 4000, - outputCap: 1000, - }, - } - restTestState.usageReplies = [ - { ...pricedReply, usage: null }, - ...Array.from({ length: 5000 }, () => ({ - ...pricedReply, - usage: { promptTokens: 1, completionTokens: 0 }, - })), - { - ...pricedReply, - providerId: restOtherId, - configuration: { - ...pricedReply.configuration, - prices: { inputPerMillion: "7", outputPerMillion: "15" }, - }, - }, - { ...pricedReply, tenantUserId: restOtherId, usage: null }, - { ...pricedReply, tenantUserId: restOtherId }, - ] + test("usage returns recorded totals and zero for users without costs", async () => { restTestState.usageTotals = [ - { tenantUserId: restUserId, cost: "1.484567", needsPricing: true }, - { tenantUserId: restOtherId, cost: "0", needsPricing: true }, + { tenantUserId: restUserId, cost: "1.484567" }, + { tenantUserId: restOtherId, cost: "0" }, ] const ids = [restUserId, restOtherId, restTenantId] restTestState.rows.push( @@ -1520,7 +1482,7 @@ describe("REST API through the Effect Fetch handler", () => { end: "2024-06-01T07:04:56.123Z", items: ids.map((id, index) => ({ tenant_user_id: id, - cost_usd: ["1.506567", "0.003", "0"][index], + cost_usd: ["1.484567", "0", "0"][index], })), }) expect(restLastPredicate().params).toEqual([restOrgId, restTenantId, ...ids, restOrgId]) @@ -1530,15 +1492,13 @@ describe("REST API through the Effect Fetch handler", () => { restTenantId, "assistant", "draft", + ...ids, + "2024-02-29T07:04:56.123Z", + "2024-06-01T07:04:56.123Z", "2024-02-29T07:04:56.123Z", "2024-06-01T07:04:56.123Z", - ...ids, ]) - expect(savedReplies.sql).toContain('"chat_message"."created_at" >=') - expect(savedReplies.sql).toContain('"chat_message"."created_at" <') - expect(new PgDialect().sqlToQuery(restTestState.usagePredicates[2]!).params).toContain( - pricedReply.createdAt, - ) + expect(restTestState.usagePredicates).toHaveLength(1) restTestState.rows.push( [restTenantRow], ids.map((id) => ({ id })), diff --git a/platform/src/routes/docs/-content/dashboard/models.md b/platform/src/routes/docs/-content/dashboard/models.md index 11b0f35e..c7d71586 100644 --- a/platform/src/routes/docs/-content/dashboard/models.md +++ b/platform/src/routes/docs/-content/dashboard/models.md @@ -17,7 +17,11 @@ Models connect your organization's agents to model providers. Let's add a provid Test results apply to the selected model and saved configuration at that moment. Editing the configuration or choosing another model clears the result. Each organization can run up to five tests per minute, and each request stops after 30 seconds. -Completing deployment configuration starts the first catalog download, then catalog defaults refresh daily from Pydantic's genai-prices catalog. A failed refresh keeps the last valid defaults. Overrides belong to this model on this connection and do not change during catalog refreshes. Turning **Override catalog defaults** off discards your overrides and restores automatic updates to prices and token limits. Context pricing tiers apply to catalog prices, while overrides use the flat rates you enter. Expand **Cache pricing** to edit cache rates. +Completing deployment configuration starts the first catalog download, then catalog defaults refresh daily from Pydantic's genai-prices catalog. A failed refresh keeps the last valid defaults. Manual prices belong to this model on this connection and do not change during catalog refreshes. Token limits continue following the catalog unless you edit them. Turning **Override catalog defaults** off discards your overrides and restores automatic updates to prices and token limits. Context pricing tiers apply to catalog prices, while overrides use the flat rates you enter. Expand **Cache pricing** to edit cache rates. + +Price changes take effect when you save. Under **Price schedule**, choose **Schedule price change**, enter its future **Effective from** time in your local time zone, set its prices, then choose **Save provider**. Expand a future entry to edit it, or choose **Cancel scheduled price** and save to remove it. Turning off overrides also cancels scheduled prices. Active versions remain immutable, and the next model request uses the rates effective at its start, including requests within an ongoing conversation. Token limits are separate and are not scheduled. + +Each model request saves its resolved rates before dispatch. A catalog refresh or later manual edit cannot change that snapshot. Existing manual prices begin their history when this feature's migration runs, without inventing prices for earlier dates. Received provider charges take precedence over token calculations. Recovery of partial usage after interruption remains separate work. These settings record prices and token bounds. When the context window includes both input and output, the configured output cap reduces the displayed input allowance. Saving validates those bounds, but chat does not yet enforce the configured input allowance or output cap. Reported model usage and prices do feed saved conversation costs and [Usage limits](./usage-limits.md). diff --git a/platform/src/routes/docs/-content/sdk/listings.md b/platform/src/routes/docs/-content/sdk/listings.md index 2a0351f0..c1c5cdd5 100644 --- a/platform/src/routes/docs/-content/sdk/listings.md +++ b/platform/src/routes/docs/-content/sdk/listings.md @@ -241,7 +241,7 @@ Conversation views also expose `directory-transcript`. The directory styling slo Organization owners and developers see **Usage (USD)** in the user directory, independently of `showAdmin`. Usage loads separately, so users and pagination remain available while costs load or fail. Choose this month, previous month, this week, previous week, last 30 days, or last 7 days. Calendar periods use UTC, with weeks running Monday through Sunday. Rolling periods end when fetched and span exactly 30 or 7 days. The current month is the default. Changing the period reloads only usage and preserves the directory state. Results are reused for 30 seconds, and Refresh reloads both users and usage with a new rolling cutoff. Viewers and Tenant administrators do not request usage. -Usage totals include retained settled assistant replies. New replies save their resolved USD cost, preferring provider-reported cost and otherwise recording the configured token cost at generation time. Older replies without a saved cost use current configured rates and can change when rates change. Drafts are excluded. Users without settled replies have zero usage. Missing settled usage or pricing contributes $0 for now, including interrupted replies whose cost is unknown. A failed usage request shows **Unavailable** with Retry. Deleted conversations are excluded. Amounts show their precise value directly, with trailing zeros omitted. Server integrations can use `getTenantUserUsage` with an Organization API key, a comma-separated `user_ids` list, and ISO 8601 `start` and `end` timestamps with an explicit timezone, such as `2026-10-01T00:00:00.000Z`. The start is inclusive and the end is exclusive. Requests support up to 100 users and any positive time range no longer than 93 days, with millisecond precision. +Usage totals include retained settled assistant replies. New requests save their resolved USD cost, preferring provider-reported cost and otherwise using rates frozen before dispatch. Each request counts in the period when it started, including retries and continuations. Older replies retain their saved aggregate cost in the period when the reply was created. Historical usage is never repriced using today's rates. Drafts are excluded. Users without settled replies have zero usage. Missing settled usage or pricing contributes $0 for now, including interrupted replies whose cost is unknown. A failed usage request shows **Unavailable** with Retry. Deleted conversations are excluded. Amounts show their precise value directly, with trailing zeros omitted. Server integrations can use `getTenantUserUsage` with an Organization API key, a comma-separated `user_ids` list, and ISO 8601 `start` and `end` timestamps with an explicit timezone, such as `2026-10-01T00:00:00.000Z`. The start is inclusive and the end is exclusive. Requests support up to 100 users and any positive time range no longer than 93 days, with millisecond precision. **NOTE**: For uncommon internal-management use cases, all three directories also support `scope="organization"` with an [organization-management token](./api.md). The user directory then offers a searchable Tenant picker. Set `tenantId` or `tenantExternalId` to pin one Tenant, or when using that token with tenant scope. Tokens retain their delegated permissions even though the UI is read-only.