Normalize historical OpenAI token layouts before report aggregation

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed 2026-10-04 01:03:53 -07:00
1 parent 236c8adee4
commit 1b790095ed
4 files changed
+67 -14

No files matched your search

@@ -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 };
+4 -2
View File
@@ -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,
+19 -11
View File
@@ -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<number>`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<typeof sql>) {
return sql<number>`coalesce(sum(${column}), 0)::double precision`;
}
@@ -305,7 +313,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
issueCount: sql<number>`count(distinct ${issues.id})::int`,
costCents: sumAsNumber(costEvents.costCents),
costCentsExact: sql<string>`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<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
costCents: sumAsNumber(costEvents.costCents),
costCentsExact: sql<string>`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<number>`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<number>`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<number>`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<number>`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<string>`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<number>`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<number>`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<number>`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<number>`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<string>`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<number>`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<number>`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<number>`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<number>`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<number>`count(distinct ${costEvents.provider})::int`,
@@ -476,7 +484,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
biller: sql<string>`case when count(distinct ${costEvents.biller}) = 1 then min(${costEvents.biller}) else 'mixed' end`,
costCents: sumAsNumber(costEvents.costCents),
costCentsExact: sql<string>`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<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
costCents: sumAsNumber(costEvents.costCents),
costCentsExact: sql<string>`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<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
inputTokens: sumAsNumber(costEvents.inputTokens),
inputTokens: sumAsNumber(ordinaryInputTokens),
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
outputTokens: sumAsNumber(costEvents.outputTokens),
})