import { receiptFingerprint } from "./receipt-fingerprint.js"; import { agentAvatarUrl, resolveAgentAppearance, createCostEventSchema, createServiceCostEventSchema, normalizeCents, isUuidLike, type MoneyInput, type CostByUserReport } from "@paperclipai/shared"; import { and, desc, eq, gte, inArray, isNotNull, isNull, lt, lte, ne, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { activityLog, agents, agentRuntimeState, authUsers, companies, companyMemberships, costEvents, heartbeatRuns, issues, projects, goals, aiSubscriptions } from "@paperclipai/db"; import { notFound, unprocessable, conflict } from "../errors.js"; import { budgetServiceInTransaction, type BudgetServiceHooks } from "./budgets.js"; import { logActivity, type LogActivityInput, type ActivityPublication } from "./activity-log.js"; import { withAccountingReadSnapshot, withAccountingTransaction } from "./accounting-transaction.js"; import { visibleIssueCondition } from "./issue-visibility.js"; import { resolveCostDateRange, type CostDateRange } from "./cost-date-range.js"; export type { CostDateRange } from "./cost-date-range.js"; const METERED_BILLING_TYPE = "metered_api"; const SUBSCRIPTION_BILLING_TYPES = ["subscription_included", "subscription_overage"] as const; // Pre-receipt Codex events stored cache reads inside input; Pi/OpenCode used // provider-qualified model IDs and already excluded cache reads. Normalize each // historical row before aggregation so mixed old/new groups share the current // exclusive-input contract. Preserve the original ledger and monetary amounts. export const ordinaryInputTokens = sql`case when ${costEvents.receiptHash} is null and ${costEvents.provider} = 'openai' and ${costEvents.model} not like '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`; } function currentUtcMonthWindow(now = new Date()) { const year = now.getUTCFullYear(); const month = now.getUTCMonth(); return { start: new Date(Date.UTC(year, month, 1, 0, 0, 0, 0)), end: new Date(Date.UTC(year, month + 1, 1, 0, 0, 0, 0)), }; } /** Increment an initialized month under the company lock. On upgrade or UTC * rollover, initialize from the ledger in the same transaction exactly once. * Reports, budgets and the integrity checker continue reading the ledger. */ export async function updateMonthlySpendProjections(db: Db, companyId: string, agentId: string | null, amountCents: MoneyInput, occurredAt: Date) { const { start, end } = currentUtcMonthWindow(); const month = start.toISOString().slice(0, 10); const delta = occurredAt >= start && occurredAt < end ? normalizeCents(amountCents) : "0.0000000"; for (const table of [companies, agents] as const) { const id = table === companies ? companyId : agentId; if (!id) continue; const companyCondition = table === companies ? eq(companies.id, companyId) : eq(agents.companyId, companyId); await db.update(table).set({ spentMonthlyCents: sql`case when ${table.spendMonthUtc} = ${month} then ${table.spentMonthlyCents} + ${delta}::numeric else ( select coalesce(sum(${costEvents.costCents}),0) from ${costEvents} where ${costEvents.companyId} = ${companyId} and ${costEvents.occurredAt} >= ${start.toISOString()}::timestamptz and ${costEvents.occurredAt} < ${end.toISOString()}::timestamptz ${table === agents ? sql`and ${costEvents.agentId} = ${agentId}` : sql``}) end`, spendMonthUtc: month, updatedAt: new Date(), }).where(and(eq(table.id, id), companyCondition)); } } export async function getMonthlySpendTotal( db: Db, scope: { companyId: string; agentId?: string | null }, ) { const { start, end } = currentUtcMonthWindow(); const conditions = [ eq(costEvents.companyId, scope.companyId), gte(costEvents.occurredAt, start), lt(costEvents.occurredAt, end), ]; if (scope.agentId) { conditions.push(eq(costEvents.agentId, scope.agentId)); } const [row] = await db .select({ total: sumAsNumber(costEvents.costCents), }) .from(costEvents) .where(and(...conditions)); return Number(row?.total ?? 0); } export async function createCostEventInTransaction(db: Db, companyId: string, data: Omit & { costCents: MoneyInput }, publications: ActivityPublication[] = [], actor?: Pick, options: { deferBudgetEvaluation?: boolean } = {}) { const parsed = ((data.usageKind === "decision" || data.usageKind === "fast_response" || data.usageKind === "prompt_injection") ? createServiceCostEventSchema : createCostEventSchema).safeParse({ ...data, occurredAt: data.occurredAt.toISOString() }); if (!parsed.success) throw unprocessable("Invalid cost receipt", parsed.error.flatten()); const values = { ...parsed.data, issueId: parsed.data.issueId ?? null, projectId: parsed.data.projectId ?? null, goalId: parsed.data.goalId ?? null, heartbeatRunId: parsed.data.heartbeatRunId ?? null, billingCode: parsed.data.billingCode ?? null, idempotencyKey: parsed.data.idempotencyKey ?? null, costCents: normalizeCents(parsed.data.costCents), occurredAt: new Date(parsed.data.occurredAt), }; const receiptHash = receiptFingerprint(values); // Older receipts hashed the numeric API value. Retain replay compatibility // only when the decimal value survives that representation exactly. let legacyHash: string | null = null; try { if (normalizeCents(Number(values.costCents)) === values.costCents) legacyHash = receiptFingerprint({ ...values, costCents: Number(values.costCents) }); } catch { /* The legacy number may round beyond the decimal storage limit. */ } if (values.idempotencyKey) { const [existing] = await db.select().from(costEvents).where(and( eq(costEvents.companyId, companyId), eq(costEvents.idempotencyKey, values.idempotencyKey), )); if (existing) { if (existing.receiptHash !== receiptHash && existing.receiptHash !== legacyHash) throw conflict("Idempotency key already used for a different cost receipt"); const [exact] = await db.select({ costCents: sql`${costEvents.costCents}::text` }).from(costEvents).where(eq(costEvents.id, existing.id)); return { ...existing, costCentsExact: normalizeCents(exact.costCents) }; } } const links = [ [agents, values.agentId, "Agent"], [issues, values.issueId, "Issue"], [projects, values.projectId, "Project"], [goals, values.goalId, "Goal"], [heartbeatRuns, values.heartbeatRunId, "Heartbeat run"], ] as const; for (const [table, id, label] of links) { if (!id) continue; const [row] = await db.select({ companyId: table.companyId }).from(table).where(eq(table.id, id)); if (!row) throw notFound(`${label} not found`); if (row.companyId !== companyId) throw unprocessable(`${label} does not belong to company`); } let subscriptionId: string | null = null; if (values.heartbeatRunId) { const [run] = await db.select({ agentId: heartbeatRuns.agentId, context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(eq(heartbeatRuns.id, values.heartbeatRunId)); if (run.agentId !== values.agentId) throw unprocessable("Heartbeat run does not belong to agent"); const attribution = run.context?.aiConnection as { subscriptionId?: unknown; provider?: unknown } | undefined; if (SUBSCRIPTION_BILLING_TYPES.includes(values.billingType as typeof SUBSCRIPTION_BILLING_TYPES[number]) && attribution?.provider === values.provider && typeof attribution.subscriptionId === "string" && isUuidLike(attribution.subscriptionId)) { const [account] = await db.select({ id: aiSubscriptions.id }).from(aiSubscriptions).where(and(eq(aiSubscriptions.id, attribution.subscriptionId), eq(aiSubscriptions.companyId, companyId), eq(aiSubscriptions.provider, values.provider))); subscriptionId = account?.id ?? null; } } // Server-derived attribution stays outside the immutable receipt fingerprint; // replaying a receipt must not depend on later account metadata changes. const [event] = await db.insert(costEvents).values({ ...values, subscriptionId, costCents: sql`${values.costCents}::numeric`, reportedCostCents: values.costCents, id: data.id, companyId, receiptHash }).returning(); await updateMonthlySpendProjections(db, companyId, event.agentId, values.costCents, event.occurredAt); // Separately reported charges linked to an already-accounted run contribute // to lifetime totals too. Before acknowledgement, accountRunCost includes // all of the run's events atomically when it initializes the projection. if (event.heartbeatRunId && event.agentId) { const [run] = await db.select({ acknowledged: heartbeatRuns.costAccountedAt, version: heartbeatRuns.accountingProjectionVersion }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, event.heartbeatRunId), eq(heartbeatRuns.companyId, companyId))); if (run.acknowledged && run.version === "v2") await db.update(agentRuntimeState).set({ totalCostCents: sql`${agentRuntimeState.totalCostCents} + ${values.costCents}::numeric`, totalInputTokens: sql`${agentRuntimeState.totalInputTokens} + ${event.inputTokens}`, totalCachedInputTokens: sql`${agentRuntimeState.totalCachedInputTokens} + ${event.cachedInputTokens}`, totalOutputTokens: sql`${agentRuntimeState.totalOutputTokens} + ${event.outputTokens}`, updatedAt: new Date(), }).where(and(eq(agentRuntimeState.agentId, event.agentId), eq(agentRuntimeState.companyId, companyId))); } // A multi-model run receipt inserts all parts before one scope evaluation. // The owning accounting transaction must perform that deferred evaluation. if (!options.deferBudgetEvaluation) await budgetServiceInTransaction(db, publications).evaluateCostEvent(event); await logActivity(db, { companyId, actorType: actor?.actorType ?? "system", actorId: actor?.actorId ?? "cost_accounting", agentId: actor?.agentId ?? event.agentId, ...((event.usageKind === "decision" || event.usageKind === "fast_response" || event.usageKind === "prompt_injection") ? { responsibleUserIdOverride: event.responsibleUserId } : {}), runId: event.heartbeatRunId, action: "cost.reported", entityType: "cost_event", entityId: event.id, details: { costCents: event.costCents, costCentsExact: values.costCents, model: event.model, costStatus: event.costStatus }, }, publications); return { ...event, costCentsExact: values.costCents }; } export function costService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { return { createEvent: async (companyId: string, data: Omit & { costCents: MoneyInput }, actor?: Pick) => { if (data.idempotencyKey?.startsWith("heartbeat:")) throw unprocessable("The heartbeat idempotency namespace is reserved for run receipts"); const event = await withAccountingTransaction(db, companyId, (tx, publications) => createCostEventInTransaction(tx, companyId, data, publications, actor)); return event; }, summary: async (companyId: string, range?: CostDateRange) => withAccountingReadSnapshot(db, companyId, async (db) => { const company = await db .select() .from(companies) .where(eq(companies.id, companyId)) .then((rows) => rows[0] ?? null); if (!company) throw notFound("Company not found"); range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); const [{ total, totalExact, eventCount, unpricedEventCount, estimatedEventCount, unknownBillingEventCount }] = await db .select({ eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, unknownBillingEventCount: sql`count(*) filter (where ${costEvents.billingType} = 'unknown')::int`, unpricedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`, total: sumAsNumber(costEvents.costCents), totalExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, }) .from(costEvents) .where(and(...conditions)); const pendingConditions = [eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.costAccountingPending, true), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])]; if (range?.from) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) >= ${range.from.toISOString()}::timestamptz`); if (range?.to) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) <= ${range.to.toISOString()}::timestamptz`); const [pending] = await db.select({ count: sql`count(*)::int` }).from(heartbeatRuns).where(and(...pendingConditions)); const pendingRunCount = pending?.count ?? 0; const spendCents = Number(total); const utilization = company.budgetMonthlyCents > 0 ? (spendCents / company.budgetMonthlyCents) * 100 : 0; return { companyId, eventCount, unpricedEventCount, unknownBillingEventCount, estimatedEventCount, pendingRunCount, pricingComplete: unpricedEventCount === 0 && pendingRunCount === 0, spendCents, spendCentsExact: normalizeCents(totalExact), budgetCents: company.budgetMonthlyCents, utilizationPercent: Number(utilization.toFixed(2)), }; }), issueTreeSummary: async ( companyId: string, issueId: string, options: { excludeRoot?: boolean } = {}, ) => { // Callers must resolve and authorize a visible root issue before invoking this. // The route does that so zero counts are not mistaken for a missing root. const childIssues = alias(issues, "child"); // The seed of the recursive CTE: when excludeRoot is true, start from // the direct children so the root issue itself is not counted. const cteSeed = options.excludeRoot ? sql` SELECT ${issues.id} FROM ${issues} WHERE ${issues.companyId} = ${companyId} AND ${issues.parentId} = ${issueId} AND ${issues.hiddenAt} IS NULL AND ${issues.harnessKind} IS NULL ` : sql` SELECT ${issues.id} FROM ${issues} WHERE ${issues.companyId} = ${companyId} AND ${issues.id} = ${issueId} AND ${issues.hiddenAt} IS NULL AND ${issues.harnessKind} IS NULL `; const cteSeedText = options.excludeRoot ? sql` SELECT (${issues.id})::text AS id FROM ${issues} WHERE ${issues.companyId} = ${companyId} AND ${issues.parentId} = ${issueId} AND ${issues.hiddenAt} IS NULL AND ${issues.harnessKind} IS NULL ` : sql` SELECT (${issues.id})::text AS id FROM ${issues} WHERE ${issues.companyId} = ${companyId} AND ${issues.id} = ${issueId} AND ${issues.hiddenAt} IS NULL AND ${issues.harnessKind} IS NULL `; const issueTreeCondition = sql` ${issues.id} IN ( WITH RECURSIVE issue_tree(id) AS ( ${cteSeed} UNION ALL SELECT ${childIssues.id} FROM ${issues} ${childIssues} JOIN issue_tree ON ${childIssues.parentId} = issue_tree.id WHERE ${childIssues.companyId} = ${companyId} AND ${childIssues.hiddenAt} IS NULL AND ${childIssues.harnessKind} IS NULL ) SELECT id FROM issue_tree ) `; const runSummarySql = sql` WITH RECURSIVE issue_tree(id) AS ( ${cteSeedText} UNION ALL SELECT (${childIssues.id})::text FROM ${issues} ${childIssues} JOIN issue_tree ON (${childIssues.parentId})::text = issue_tree.id WHERE ${childIssues.companyId} = ${companyId} AND ${childIssues.hiddenAt} IS NULL AND ${childIssues.harnessKind} IS NULL ) SELECT count(distinct ${heartbeatRuns.id})::int AS "runCount", coalesce(sum(extract(epoch from (coalesce(${heartbeatRuns.finishedAt}, now()) - ${heartbeatRuns.startedAt})) * 1000), 0)::double precision AS "runtimeMs" FROM ${heartbeatRuns} WHERE ${heartbeatRuns.companyId} = ${companyId} AND ${heartbeatRuns.startedAt} IS NOT NULL AND ( ${heartbeatRuns.contextSnapshot} ->> 'issueId' IN (SELECT id FROM issue_tree) OR EXISTS ( SELECT 1 FROM ${activityLog} JOIN issue_tree ON ${activityLog.entityId} = issue_tree.id WHERE ${activityLog.companyId} = ${companyId} AND ${activityLog.entityType} = 'issue' AND ${activityLog.runId} = ${heartbeatRuns.id} ) ) `; // Run cost-event aggregation and run-duration aggregation in parallel. // They're separate queries because cost_events fan out per-event and // joining heartbeat_runs through them would double-count run durations. const [costRowResult, runRowResult] = await Promise.all([ db .select({ issueCount: sql`count(distinct ${issues.id})::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) .from(issues) .leftJoin( costEvents, and( eq(costEvents.companyId, companyId), eq(costEvents.issueId, issues.id), ), ) .where( and( eq(issues.companyId, companyId), visibleIssueCondition(), issueTreeCondition, ), ), db.execute(runSummarySql), ]); const costRow = costRowResult[0]; const runRow = Array.isArray(runRowResult) ? (runRowResult[0] as { runCount?: number | string | null; runtimeMs?: number | string | null } | undefined) : undefined; return { issueId, issueCount: Number(costRow?.issueCount ?? 0), includeDescendants: true, costCents: Number(costRow?.costCents ?? 0), costCentsExact: normalizeCents(costRow?.costCentsExact ?? 0), inputTokens: Number(costRow?.inputTokens ?? 0), cachedInputTokens: Number(costRow?.cachedInputTokens ?? 0), outputTokens: Number(costRow?.outputTokens ?? 0), runCount: Number(runRow?.runCount ?? 0), runtimeMs: Number(runRow?.runtimeMs ?? 0), }; }, byUser: (companyId: string, range?: CostDateRange): Promise => withAccountingReadSnapshot(db, companyId, async tx => { // Keep membership, identity and spend in one non-blocking read snapshot. const members = await tx.select({ userId: companyMemberships.principalId, status: companyMemberships.status, userName: authUsers.name, userImage: authUsers.image, }).from(companyMemberships) .leftJoin(authUsers, eq(authUsers.id, companyMemberships.principalId)) .where(and( eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalType, "user"), ne(companyMemberships.principalId, "local-board"), )); const activeMembers = members.filter(member => member.status === "active"); const activeUserCount = activeMembers.length; const bounds = resolveCostDateRange(range); const conditions = [eq(costEvents.companyId, companyId)]; if (bounds.from) conditions.push(gte(costEvents.occurredAt, bounds.from)); if (bounds.to) conditions.push(lte(costEvents.occurredAt, bounds.to)); const totals = await tx.select({ userId: companyMemberships.principalId, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, unpricedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), runCount: sql`count(distinct ${heartbeatRuns.id})::int`, }).from(costEvents) .leftJoin(heartbeatRuns, and( eq(heartbeatRuns.id, costEvents.heartbeatRunId), eq(heartbeatRuns.companyId, costEvents.companyId), eq(heartbeatRuns.agentId, costEvents.agentId), )) // Never resolve names through a foreign-company run or membership. Retain // former members' historical spend; missing attribution remains in the total. // The synthetic Board principal is not a person. Any historical charges // attached to it remain in Unattributed so the report still reconciles. .leftJoin(companyMemberships, and( eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalType, "user"), ne(companyMemberships.principalId, "local-board"), eq(companyMemberships.principalId, sql`case when ${costEvents.usageKind} in ('decision', 'fast_response', 'prompt_injection') then ${costEvents.responsibleUserId} else ${heartbeatRuns.responsibleUserId} end`), )) .where(and(...conditions)) .groupBy(companyMemberships.principalId) .orderBy(desc(sql`sum(${costEvents.costCents})`), companyMemberships.principalId); const directory = new Map(members.map(member => [member.userId, member])); const rows: CostByUserReport["rows"] = totals.map(row => ({ ...row, userName: row.userId ? directory.get(row.userId)?.userName ?? "Former user" : null, userImage: row.userId ? directory.get(row.userId)?.userImage ?? null : null, costCentsExact: normalizeCents(row.costCentsExact), })); const represented = new Set(rows.map(row => row.userId)); for (const member of activeMembers.sort((a, b) => (a.userName ?? a.userId).localeCompare(b.userName ?? b.userId))) { if (represented.has(member.userId)) continue; rows.push({ userId: member.userId, userName: member.userName ?? "Unknown user", userImage: member.userImage, eventCount: 0, estimatedEventCount: 0, unpricedEventCount: 0, costCents: 0, costCentsExact: "0.0000000", inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, runCount: 0, }); } return { activeUserCount, rows }; }), byAgent: async (companyId: string, range?: CostDateRange) => { range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); const rows = await db .select({ agentId: costEvents.agentId, agentName: agents.name, agentAppearance: agents.appearance, agentStatus: agents.status, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: sql`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`, subscriptionRunCount: sql`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`, 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 ${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`, }) .from(costEvents) .leftJoin(agents, eq(costEvents.agentId, agents.id)) .where(and(...conditions)) .groupBy(costEvents.agentId, agents.name, agents.appearance, agents.status) .orderBy(desc(sumAsNumber(costEvents.costCents))); return rows.map(row => { const appearance = row.agentId ? resolveAgentAppearance(row.agentAppearance, row.agentId) : null; return { ...row, agentAppearance: appearance, avatarUrl: appearance ? agentAvatarUrl(appearance, 512) : undefined }; }); }, byProvider: async (companyId: string, range?: CostDateRange) => { range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); return db .select({ provider: costEvents.provider, biller: costEvents.biller, billingType: costEvents.billingType, model: costEvents.model, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, unpricedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: sql`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`, subscriptionRunCount: sql`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`, 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 ${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`, }) .from(costEvents) .where(and(...conditions)) .groupBy(costEvents.provider, costEvents.biller, costEvents.billingType, costEvents.model) .orderBy(desc(sumAsNumber(costEvents.costCents))); }, byBiller: async (companyId: string, range?: CostDateRange) => { range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); return db .select({ biller: costEvents.biller, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, unpricedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), apiRunCount: sql`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`, subscriptionRunCount: sql`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`, 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 ${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`, modelCount: sql`count(distinct ${costEvents.model})::int`, }) .from(costEvents) .where(and(...conditions)) .groupBy(costEvents.biller) .orderBy(desc(sumAsNumber(costEvents.costCents))); }, /** * aggregates cost_events by provider for each of three rolling windows: * last 5 hours, last 24 hours, last 7 days. * purely internal consumption data, no external rate-limit sources. */ windowSpend: async (companyId: string) => { const windows = [ { label: "5h", hours: 5 }, { label: "24h", hours: 24 }, { label: "7d", hours: 168 }, ] as const; const now = new Date(); const results = await Promise.all( windows.map(async ({ label, hours }) => { const since = new Date(now.getTime() - hours * 60 * 60 * 1000); const rows = await db .select({ provider: costEvents.provider, 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(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) .from(costEvents) .where( and( eq(costEvents.companyId, companyId), gte(costEvents.occurredAt, since), lte(costEvents.occurredAt, now), ), ) .groupBy(costEvents.provider) .orderBy(desc(sumAsNumber(costEvents.costCents))); return rows.map((row) => ({ provider: row.provider, biller: row.biller, window: label as string, windowHours: hours, costCents: row.costCents, costCentsExact: normalizeCents(row.costCentsExact), inputTokens: row.inputTokens, cachedInputTokens: row.cachedInputTokens, outputTokens: row.outputTokens, })); }), ); return results.flat(); }, byAgentModel: async (companyId: string, range?: CostDateRange) => { range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); // single query: group by agent + provider + model. // the (companyId, agentId, occurredAt) composite index covers this well. // order by provider + model for stable db-level ordering; cost-desc sort // within each agent's sub-rows is done client-side in the ui memo. const rows = await db .select({ agentId: costEvents.agentId, agentName: agents.name, agentAppearance: agents.appearance, provider: costEvents.provider, biller: costEvents.biller, billingType: costEvents.billingType, model: costEvents.model, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, costCents: sumAsNumber(costEvents.costCents), costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) .from(costEvents) .leftJoin(agents, eq(costEvents.agentId, agents.id)) .where(and(...conditions)) .groupBy( costEvents.agentId, agents.name, agents.appearance, costEvents.provider, costEvents.biller, costEvents.billingType, costEvents.model, ) .orderBy(costEvents.provider, costEvents.biller, costEvents.billingType, costEvents.model); return rows.map(row => { const appearance = row.agentId ? resolveAgentAppearance(row.agentAppearance, row.agentId) : null; return { ...row, agentAppearance: appearance, avatarUrl: appearance ? agentAvatarUrl(appearance, 512) : undefined }; }); }, byProject: async (companyId: string, range?: CostDateRange) => { const issueIdAsText = sql`${issues.id}::text`; const runProjectLinks = db .select({ runId: activityLog.runId, projectId: sql`min(${issues.projectId}::text)::uuid`.as("linked_project_id"), }) .from(activityLog) .innerJoin( issues, and( eq(activityLog.entityType, "issue"), eq(activityLog.entityId, issueIdAsText), ), ) .where( and( eq(activityLog.companyId, companyId), eq(issues.companyId, companyId), isNotNull(activityLog.runId), isNotNull(issues.projectId), ), ) .groupBy(activityLog.runId) .having(sql`count(distinct ${issues.projectId}) = 1`) .as("run_project_links"); const effectiveProjectId = sql`coalesce(${costEvents.projectId}, case when ${costEvents.receiptHash} is null then ${runProjectLinks.projectId} end)`; range = resolveCostDateRange(range); const conditions: ReturnType[] = [eq(costEvents.companyId, companyId)]; if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from)); if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to)); const costCentsExpr = sumAsNumber(costEvents.costCents); return db .select({ projectId: projects.id, projectName: projects.name, eventCount: sql`count(*)::int`, estimatedEventCount: sql`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`, costCents: costCentsExpr, costCentsExact: sql`coalesce(sum(${costEvents.costCents}), 0)::text`, inputTokens: sumAsNumber(ordinaryInputTokens), cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens), outputTokens: sumAsNumber(costEvents.outputTokens), }) .from(costEvents) .leftJoin(runProjectLinks, eq(costEvents.heartbeatRunId, runProjectLinks.runId)) .leftJoin(projects, and(eq(projects.companyId, companyId), sql`${projects.id} = ${effectiveProjectId}`)) .where(and(...conditions)) .groupBy(projects.id, projects.name) .orderBy(desc(costCentsExpr)); }, }; }