From 7d7e66d5da5f764d6fc6f82c34d4c8b1343f3131 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 4 Oct 2026 01:01:43 -0700 Subject: [PATCH 1/2] Defer fresh executions during credential lock contention Co-Authored-By: Paperclip --- doc/connections/AI-CONNECTIONS.md | 7 ++-- .../agent-hire-ai-connections.test.ts | 32 +++++++++++++++++++ server/src/__tests__/ai-connections.test.ts | 31 +++++++++++++++++- server/src/services/ai-connections.ts | 13 ++++++++ server/src/services/heartbeat.ts | 5 ++- 5 files changed, 81 insertions(+), 7 deletions(-) diff --git a/doc/connections/AI-CONNECTIONS.md b/doc/connections/AI-CONNECTIONS.md index c74060853f9..29bd19a37eb 100644 --- a/doc/connections/AI-CONNECTIONS.md +++ b/doc/connections/AI-CONNECTIONS.md @@ -498,9 +498,10 @@ existing account name. Connecting installs access for that agent and resumes the work automatically. Explicit incompatible bindings and shared-account permission denials still fail; hiring never expands a restricted shared account's audience. -Concurrent runs of one subscription do not wait for each other. No credential -lease exists to hold them, so a fresh task execution cannot enter a contention -wait. A run that already entered this wait keeps its scheduled retries. It does +Concurrent runs of one subscription do not hold a credential lease. A fresh +task can briefly wait when a credential rotation holds the company file lock or +a grant/secret database row lock. Lock timeouts become `ai_connection_busy`, +keeping the task on its automatic pre-provider scheduled retry path. It does not request new credentials, and it does not consume the provider-failure retry allowance. Each retry revalidates the account, and existing run-dispatch rules still suppress cancelled, reassigned, or otherwise ineligible work. An assignee diff --git a/server/src/__tests__/agent-hire-ai-connections.test.ts b/server/src/__tests__/agent-hire-ai-connections.test.ts index d4f28de63d0..5c7d4ebda7b 100644 --- a/server/src/__tests__/agent-hire-ai-connections.test.ts +++ b/server/src/__tests__/agent-hire-ai-connections.test.ts @@ -1,3 +1,5 @@ +import * as codexAdapter from "@paperclipai/adapter-codex-local/server"; +import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge"; import { randomUUID } from "node:crypto"; import { mkdtemp, rm } from "node:fs/promises"; import os from "node:os"; @@ -323,6 +325,36 @@ describe("agent-created hires use managed AI connections", () => { }); describe("hired agents sharing a subscription", () => { + it("keeps a credential-lock timeout on automatic retry without blocking the task or starting a provider", async () => { + const f = await fixture("openai", "subscription"); + const agent = hired(await request(f.app).post(`/api/companies/${f.companyId}/agent-hires`).send({ name: "Waiting teammate", role: "engineer", adapterType: f.adapterType, reportsTo: f.agentId, adapterConfig: { cwd: home, engine: "cli" }, runtimeConfig: { heartbeat: { enabled: false } } })); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Wait for credential rotation", status: "todo", assigneeAgentId: agent.id, responsibleUserId: f.userId, createdByUserId: f.userId }).returning(); + const execute = vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, resultJson: {} })); + registerServerAdapter({ ...getServerAdapter(f.adapterType), execute }); + const lock = vi.spyOn(codexAdapter, "withAccountHomeSecretMutationLock") + .mockRejectedValueOnce(Object.assign(new Error("quota refresh holds company lock"), { code: WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE })); + const heartbeat = heartbeatService(db); + try { + const run = await heartbeat.invoke(agent.id, "assignment", { issueId: issue.id, wakeReason: "issue_assigned", responsibleUserId: f.userId }, "system"); + expect(run).not.toBeNull(); + await expect.poll(async () => (await heartbeat.getRun(run!.id))?.status, { timeout: 20_000 }).toBe("cancelled"); + await heartbeat.drainActiveRunExecutions(); + expect(await heartbeat.getRun(run!.id)).toMatchObject({ errorCode: "ai_connection_busy", resultJson: { executionRecovery: { providerWorkStarted: false } } }); + const retries = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id)); + expect(retries).toHaveLength(1); + expect(retries[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "ai_connection_busy" }); + const [savedIssue] = await db.select().from(issues).where(eq(issues.id, issue.id)); + expect(savedIssue.status).not.toBe("blocked"); + expect(savedIssue.executionRunId).toBe(retries[0].id); + expect(execute).not.toHaveBeenCalled(); + expect(await aiConnectionService(db).list(f.companyId, f.userId)).toEqual([expect.objectContaining({ status: "connected" })]); + } finally { + lock.mockRestore(); + await heartbeat.drainActiveRunExecutions(); + unregisterServerAdapter(f.adapterType); + } + }); + it.each(["openai", "anthropic"] as const)("runs the %s child alongside a live parent and inherits its connection", async (provider) => { const f = await fixture(provider, "subscription"); const agent = hired(await request(f.app).post(`/api/companies/${f.companyId}/agent-hires`).send({ name: "Concurrent teammate", role: "engineer", adapterType: f.adapterType, reportsTo: f.agentId, adapterConfig: { cwd: home, engine: "cli" }, runtimeConfig: { heartbeat: { enabled: false } } })); diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index f2004abac4b..ee125c81aea 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -16,7 +16,7 @@ import { aiConnectionService } from "../services/ai-connections.js"; import * as codexAdapter from "@paperclipai/adapter-codex-local/server"; import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge"; import * as executionTarget from "@paperclipai/adapter-utils/execution-target"; -import { prepareManagedAiRuntime, assertManagedAiProjectAuth } from "../services/ai-connection-runtime.js"; +import { prepareManagedAiRuntime, assertManagedAiProjectAuth, isAiConnectionBusy } from "../services/ai-connection-runtime.js"; import { toolAccessService } from "../services/tool-access.js"; import { secretService } from "../services/secrets.js"; import { aiConnectionBindingSchema, connectionPurposeTransportSchema, isAiConnectionCompatible } from "@paperclipai/shared"; @@ -143,6 +143,35 @@ describe("managed AI connections", () => { } }); + it.each(["company_file", "grant_row", "secret_row"] as const)("defers runtime preparation on a contended %s lock and succeeds once it clears", async (kind) => { + const owner = `runtime-lock-${kind}`; + await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" }); + const auth = JSON.stringify({ tokens: { access_token: "valid-access", refresh_token: "valid-refresh", account_id: owner } }); + await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, auth); + const [row] = await service.quotaAccounts(companyId, owner); + const ref = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!; + const prepare = () => prepareManagedAiRuntime(db, { companyId, agentId, responsibleUserId: owner, adapterType: "codex_local", + binding: { provider: "openai", method: "subscription", mode: "responsible_user" }, config: {} }); + const held = deferredSignal(), release = deferredSignal(); + const lock = kind === "company_file" ? vi.spyOn(codexAdapter, "withAccountHomeSecretMutationLock") + .mockRejectedValueOnce(Object.assign(new Error("company lock held by another account"), { code: WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE })) : undefined; + const holding = kind === "company_file" ? Promise.resolve() : db.transaction(async tx => { + if (kind === "grant_row") await tx.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)).for("update"); + else await tx.select().from(companySecrets).where(eq(companySecrets.id, ref.secretId)).for("update"); + held.resolve(); await release.promise; + }); + try { + if (kind !== "company_file") await held.promise; + const error = await prepare().then(() => { throw new Error("Runtime must defer"); }, error => error); + expect(isAiConnectionBusy(error)).toBe(true); + expect(error).toMatchObject({ status: 422, details: { code: "ai_connection_busy" } }); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)))[0].status).toBe("active"); + } finally { release.resolve(); await holding; lock?.mockRestore(); } + const run = await prepare(); + try { expect(await readFile(path.join(String(run.config.env.CODEX_HOME), "auth.json"), "utf8")).toBe(auth); } + finally { await run.cleanup(); } + }); + it.each(["runtime", "failed_run"] as const)("recovers pending quota credentials before %s can reuse or invalidate the consumed token", async (consumer) => { const owner = `quota-consumer-${consumer}`; await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" }); diff --git a/server/src/services/ai-connections.ts b/server/src/services/ai-connections.ts index 323dd4652db..7e53fc78d31 100644 --- a/server/src/services/ai-connections.ts +++ b/server/src/services/ai-connections.ts @@ -1,3 +1,4 @@ +import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge"; import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server"; import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js"; import { createHash, randomUUID } from "node:crypto"; @@ -861,6 +862,18 @@ export function aiConnectionService(db: Db) { if (recovered) await quotaCredentialRecovery(row.connection.companyId, row.grant.id).clear() .catch(error => logger.warn({ err: error, grantId: row.grant.id }, "Saved runtime credential recovery cleanup pending")); return value; + }).catch(error => { + // Drizzle wraps Postgres errors in `cause`. Contention is a temporary + // pre-provider wait, not evidence that this account needs reconnecting. + let current: unknown = error; + for (let depth = 0; depth < 4 && current && typeof current === "object"; depth += 1) { + const value = current as { code?: unknown; cause?: unknown }; + if (value.code === WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE || value.code === "55P03") { + throw unprocessable("AI credentials are being updated. This execution will retry automatically.", { code: "ai_connection_busy" }); + } + current = value.cause; + } + throw error; }); } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4c7f3a281c6..1f165c3692c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -16361,9 +16361,8 @@ export function heartbeatService( .then((rows) => rows[0] ?? null); } - // No code path raises the `ai_connection_busy` error any more. This - // function stays because a stored run row can still carry that error code - // from an earlier release. + // Credential rotation can briefly contend with a fresh runtime read. Keep + // the task on its automatic pre-provider retry path while the lock clears. async function finalizeAiConnectionBusyDeferral( run: typeof heartbeatRuns.$inferSelect, error: HttpError, From 1b790095edea509cbf94d92b19c2faa91b53a12c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 4 Oct 2026 01:03:53 -0700 Subject: [PATCH 2/2] Normalize historical OpenAI token layouts before report aggregation Co-Authored-By: Paperclip --- doc/COST-ACCOUNTING.md | 2 +- .../cost-accounting-reliability.test.ts | 43 +++++++++++++++++++ server/src/__tests__/costs-service.test.ts | 6 ++- server/src/services/costs.ts | 30 ++++++++----- 4 files changed, 67 insertions(+), 14 deletions(-) diff --git a/doc/COST-ACCOUNTING.md b/doc/COST-ACCOUNTING.md index 41cdb25c261..f613cc87565 100644 --- a/doc/COST-ACCOUNTING.md +++ b/doc/COST-ACCOUNTING.md @@ -42,7 +42,7 @@ Reservations are estimates, not provider-enforced spending caps. Delayed receipt Cost and finance endpoints default to the current UTC month through now. Explicit `from`/`to` bounds remain inclusive; `period=all` requests all time. Reversed or malformed bounds are rejected. CLI cost/finance reads expose `--from`, `--to`, and `--all-time`. The board sends explicit ranges, and month/year-to-date starts use UTC. Rolling spend excludes future events. -Cost summaries expose `eventCount`, `estimatedEventCount`, `unpricedEventCount`, `pendingRunCount`, and `pricingComplete`. Estimated charges count toward budgets, while the UI distinguishes estimates from provider-reported dollars. Agent totals and expanded model rows show **Estimated** when every charge in that group is estimated, or **Partially estimated** for a mix. Groups with no estimates have no estimate badge. These labels follow the selected date range and use ledger-event counts, not run counts; they do not infer pricing status from the provider or model name. `pricingComplete` means no missing prices or pending receipts; it does not mean invoice reconciliation has occurred. The Costs page warns when totals are incomplete. Provider cards include cached tokens and calculate subscription shares using each token once. Provider spend is compared with the real company budget, not a fabricated proportional allocation. Run displays prefer cache-adjusted prices and preserve explicit zeroes. +Cost summaries expose `eventCount`, `estimatedEventCount`, `unpricedEventCount`, `pendingRunCount`, and `pricingComplete`. Estimated charges count toward budgets, while the UI distinguishes estimates from provider-reported dollars. Agent totals and expanded model rows show **Estimated** when every charge in that group is estimated, or **Partially estimated** for a mix. Groups with no estimates have no estimate badge. These labels follow the selected date range and use ledger-event counts, not run counts; they do not infer pricing status from the provider or model name. `pricingComplete` means no missing prices or pending receipts; it does not mean invoice reconciliation has occurred. The Costs page warns when totals are incomplete. Provider cards include cached tokens and calculate subscription shares using each token once. Reports normalize historical OpenAI events without a receipt fingerprint from inclusive input to exclusive input before grouping, so mixed historical/new totals count cache reads once across agents, models, providers, billers, projects, issues, and rolling windows. This read-time conversion preserves original ledger rows and charges; legacy Anthropic input already excludes cache reads. Provider spend is compared with the real company budget, not a fabricated proportional allocation. Run displays prefer cache-adjusted prices and preserve explicit zeroes. Gemini token normalization follows the [CLI stream statistics](https://github.com/google-gemini/gemini-cli/blob/main/packages/core/src/output/stream-json-formatter.ts) and [API usage metadata](https://ai.google.dev/api/generate-content#UsageMetadata): cached prompt tokens are separated from input, and thinking tokens contribute to output. diff --git a/server/src/__tests__/cost-accounting-reliability.test.ts b/server/src/__tests__/cost-accounting-reliability.test.ts index 3b7be23c970..987d97ea687 100644 --- a/server/src/__tests__/cost-accounting-reliability.test.ts +++ b/server/src/__tests__/cost-accounting-reliability.test.ts @@ -41,6 +41,49 @@ databaseDescribe("cost accounting reliability (PostgreSQL)", () => { return { company, agent, project, issue, goal, run, receipt }; } + it("counts cache reads once across mixed historical and receipt-backed reports without changing charges", async () => { + const f = await fixture(); + const costs = costService(db); + const scope = { agentId: f.agent.id, issueId: f.issue.id, projectId: f.project.id, heartbeatRunId: f.run.id, + provider: "openai", biller: "openai", model: "gpt-test", occurredAt: new Date() }; + // These two layouts both describe 100 input (80 cached) and 10 output. + for (const billingType of ["metered_api", "subscription_included"]) { + await db.insert(costEvents).values({ companyId: f.company.id, ...scope, billingType, + inputTokens: 100, cachedInputTokens: 80, outputTokens: 10, costCents: 7 }); + await costs.createEvent(f.company.id, { ...scope, billingType, + inputTokens: 20, cachedInputTokens: 80, outputTokens: 10, costCents: 11 }); + } + const [provider, models] = await Promise.all([costs.byProvider(f.company.id), costs.byAgentModel(f.company.id)]); + for (const rows of [provider, models]) { + expect(rows).toHaveLength(2); + for (const row of rows) { + expect(row).toMatchObject({ inputTokens: 40, cachedInputTokens: 160, outputTokens: 20, costCents: 18 }); + expect(row.inputTokens + row.cachedInputTokens + row.outputTokens).toBe(220); + } + } + const [agent] = await costs.byAgent(f.company.id); + const [biller] = await costs.byBiller(f.company.id); + const [project] = await costs.byProject(f.company.id); + const issue = await costs.issueTreeSummary(f.company.id, f.issue.id); + const windows = await costs.windowSpend(f.company.id); + expect(windows).toHaveLength(3); + for (const row of [agent, biller, project, issue, ...windows]) { + expect(row).toMatchObject({ inputTokens: 80, cachedInputTokens: 320, outputTokens: 40, costCents: 36 }); + expect(row.inputTokens + row.cachedInputTokens + row.outputTokens).toBe(440); + } + for (const row of [agent, biller, provider.find(row => row.billingType === "subscription_included")!]) { + expect(row).toMatchObject({ subscriptionInputTokens: 40, subscriptionCachedInputTokens: 160, subscriptionOutputTokens: 20 }); + } + // Reading reports never rewrites historical evidence or retroactively bills. + const stored = await db.select().from(costEvents).where(eq(costEvents.companyId, f.company.id)); + expect(stored.filter(row => row.receiptHash === null).map(row => row.inputTokens)).toEqual([100, 100]); + expect((await costs.summary(f.company.id)).spendCents).toBe(36); + // Legacy Anthropic input was already exclusive; retain its cache semantics. + await db.insert(costEvents).values({ companyId: f.company.id, ...scope, provider: "anthropic", billingType: "metered_api", + inputTokens: 100, cachedInputTokens: 80, outputTokens: 10, costCents: 0 }); + expect((await costs.byProvider(f.company.id)).find(row => row.provider === "anthropic")).toMatchObject({ inputTokens: 100, cachedInputTokens: 80, outputTokens: 10 }); + }); + it.each(["issueId", "projectId", "goalId", "heartbeatRunId"] as const)("rejects a foreign-company %s without writing anything", async (field) => { const a = await fixture(); const b = await fixture(); const foreignIds = { issueId: b.issue.id, projectId: b.project.id, goalId: b.goal.id, heartbeatRunId: b.run.id }; diff --git a/server/src/__tests__/costs-service.test.ts b/server/src/__tests__/costs-service.test.ts index 862ac09c666..0ea925ebb07 100644 --- a/server/src/__tests__/costs-service.test.ts +++ b/server/src/__tests__/costs-service.test.ts @@ -551,7 +551,9 @@ describeEmbeddedPostgres("cost and finance aggregate overflow handling", () => { const [byAgentModelRow] = await costs.byAgentModel(companyId, range); expect(byAgentRow?.costCents).toBe(4_000_000_000); - expect(byAgentRow?.inputTokens).toBe(4_000_000_000); + // Historical OpenAI input includes these ten cached tokens. + expect(byAgentRow?.inputTokens).toBe(3_999_999_990); + expect(byAgentRow!.inputTokens + byAgentRow!.cachedInputTokens).toBe(4_000_000_000); expect(byProjectRow?.costCents).toBe(4_000_000_000); expect(byAgentModelRow?.costCents).toBe(4_000_000_000); }); @@ -701,7 +703,7 @@ describeEmbeddedPostgres("cost and finance aggregate overflow handling", () => { includeDescendants: true, costCents: 600, costCentsExact: "600.0000000", - inputTokens: 60, + inputTokens: 54, cachedInputTokens: 6, outputTokens: 12, runCount: 0, diff --git a/server/src/services/costs.ts b/server/src/services/costs.ts index b25a3e3bb9d..c58106a64b1 100644 --- a/server/src/services/costs.ts +++ b/server/src/services/costs.ts @@ -16,7 +16,15 @@ export type { CostDateRange } from "./cost-date-range.js"; const METERED_BILLING_TYPE = "metered_api"; const SUBSCRIPTION_BILLING_TYPES = ["subscription_included", "subscription_overage"] as const; -function sumAsNumber(column: typeof costEvents.costCents | typeof costEvents.inputTokens | typeof costEvents.cachedInputTokens | typeof costEvents.outputTokens) { +// Pre-receipt OpenAI events stored cache reads inside input. Normalize each +// historical row before aggregation so mixed old/new groups share the current +// exclusive-input contract. Preserve the original ledger and monetary amounts. +const ordinaryInputTokens = sql`case + when ${costEvents.receiptHash} is null and ${costEvents.provider} = 'openai' + then greatest(0, ${costEvents.inputTokens} - ${costEvents.cachedInputTokens}) + else ${costEvents.inputTokens} end`; + +function sumAsNumber(column: typeof costEvents.costCents | typeof costEvents.inputTokens | typeof costEvents.cachedInputTokens | typeof costEvents.outputTokens | ReturnType) { return sql`coalesce(sum(${column}), 0)::double precision`; } @@ -305,7 +313,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { issueCount: sql`count(distinct ${issues.id})::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) @@ -362,7 +370,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: @@ -372,7 +380,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { subscriptionCachedInputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`, subscriptionInputTokens: - sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`, + sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`, subscriptionOutputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`, }) @@ -401,7 +409,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { model: costEvents.model, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: @@ -411,7 +419,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { subscriptionCachedInputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`, subscriptionInputTokens: - sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`, + sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`, subscriptionOutputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`, }) @@ -432,7 +440,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { biller: costEvents.biller, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: @@ -442,7 +450,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { subscriptionCachedInputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`, subscriptionInputTokens: - sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`, + sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`, subscriptionOutputTokens: sql`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`, providerCount: sql`count(distinct ${costEvents.provider})::int`, @@ -476,7 +484,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { biller: sql`case when count(distinct ${costEvents.biller}) = 1 then min(${costEvents.biller}) else 'mixed' end`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) @@ -531,7 +539,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) @@ -595,7 +603,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { projectName: projects.name, costCents: costCentsExpr, costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, - inputTokens: sumAsNumber(costEvents.inputTokens), + inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), })