diff --git a/apps/docs/content/docs/platform/costs.mdx b/apps/docs/content/docs/platform/costs.mdx index 9f77c9e5829..f6bebbb8de9 100644 --- a/apps/docs/content/docs/platform/costs.mdx +++ b/apps/docs/content/docs/platform/costs.mdx @@ -249,7 +249,7 @@ Execution rate buckets can burst above the sustained per-minute rate. General AP | Plan | Concurrent runs | | --- | --- | -| Free | 10 | +| Free | 2 | | Pro | 50 | | Max | 200 | | Enterprise default | 1,000; customizable | diff --git a/apps/sim/app/api/billing/update-cost/route.integration.ts b/apps/sim/app/api/billing/update-cost/route.integration.ts index 014ac9f70e6..2609beb6cf8 100644 --- a/apps/sim/app/api/billing/update-cost/route.integration.ts +++ b/apps/sim/app/api/billing/update-cost/route.integration.ts @@ -1,11 +1,12 @@ /** * Cost callbacks against real PostgreSQL: a direct-v1 run that outlives its admitted Stripe period * records its later spend in the payer's current period, so the closed period is never topped up - * after its invoice, and spend after the payer's terminal settlement is refused. Only the - * internal-key check is stubbed. + * after its invoice, and spend after the payer's terminal settlement is refused. A free payer's + * callback that takes it past its limit pauses the run even while an earlier admission is + * cached, and a paid payer keeps the cached admission. Only the internal-key check is stubbed. */ import { db } from '@sim/db' -import { subscription, usageLog, user, userStats } from '@sim/db/schema' +import { project, subscription, usageLog, user, userStats, workspace } from '@sim/db/schema' import { envFlagsMock } from '@sim/testing/mocks/env-flags.mock' import { generateId } from '@sim/utils/id' import { eq, inArray } from 'drizzle-orm' @@ -24,8 +25,12 @@ vi.mock('@/lib/mothership/request/http', async (importOriginal) => ({ import { BILLING_ACCOUNT_DECISION_HEADER, + BILLING_ATTRIBUTION_HEADER, + type BillingAttributionSnapshot, serializeAccountBillingDecisionHeader, + serializeBillingAttributionHeader, } from '@/lib/billing/core/attribution' +import { defaultBillingPeriod } from '@/lib/billing/core/billing-period' import { claimTerminalPeriod } from '@/lib/billing/cycle-close' import { POST } from '@/app/api/billing/update-cost/route' @@ -35,6 +40,8 @@ const userIds: string[] = [] afterAll(async () => { if (userIds.length === 0) return await db.delete(usageLog).where(inArray(usageLog.userId, userIds)) + await db.delete(workspace).where(inArray(workspace.ownerId, userIds)) + await db.delete(project).where(inArray(project.ownerId, userIds)) await db.delete(subscription).where(inArray(subscription.referenceId, userIds)) await db.delete(userStats).where(inArray(userStats.userId, userIds)) await db.delete(user).where(inArray(user.id, userIds)) @@ -47,10 +54,9 @@ interface Payer { decision: string } -/** A user on a pro subscription for `period`, whose close marker has caught up to it. */ -async function createPayer(period: { start: Date; end: Date }): Promise { +/** A user with the default $5 usage limit and no subscription. */ +async function createUser(): Promise { const userId = `update-cost-user-${generateId()}` - const subscriptionId = generateId() userIds.push(userId) await db.insert(user).values({ id: userId, @@ -61,6 +67,13 @@ async function createPayer(period: { start: Date; end: Date }): Promise { updatedAt: new Date(), }) await db.insert(userStats).values({ id: generateId(), userId }) + return userId +} + +/** A user on a pro subscription for `period`, whose close marker has caught up to it. */ +async function createPayer(period: { start: Date; end: Date }): Promise { + const userId = await createUser() + const subscriptionId = generateId() await db.insert(subscription).values({ id: subscriptionId, plan: 'pro', @@ -94,6 +107,21 @@ function requestRows(requestKey: string) { .where(inArray(usageLog.eventKey, [`update-cost:${requestKey}`, `update-cost:${requestKey}@1`])) } +/** A free user's direct-v1 decision, admitted in the open default period as admission mints it. */ +async function createFreePayer(): Promise { + const userId = await createUser() + const decision = serializeAccountBillingDecisionHeader({ + userId, + billingEntity: { type: 'user', id: userId }, + billingPeriod: { + start: defaultBillingPeriod().start.toISOString(), + end: defaultBillingPeriod().end.toISOString(), + source: 'default', + }, + }) + return { userId, subscriptionId: '', decision } +} + function callback(payer: Payer, requestKey: string, cost: number): NextRequest { return new NextRequest('http://localhost:3000/api/billing/update-cost', { method: 'POST', @@ -114,6 +142,65 @@ function callback(payer: Payer, requestKey: string, cost: number): NextRequest { }) } +/** A free user's personal workspace and the attribution its runs are admitted with. */ +async function createFreeWorkspacePayer(): Promise { + const userId = await createUser() + const projectId = generateId() + const workspaceId = generateId() + await db.insert(project).values({ id: projectId, name: 'Update Cost Test', ownerId: userId }) + await db.insert(workspace).values({ + id: workspaceId, + projectId, + name: 'Update Cost Test', + ownerId: userId, + billedAccountUserId: userId, + }) + return { + actorUserId: userId, + workspaceId, + organizationId: null, + billedAccountUserId: userId, + billingEntity: { type: 'user', id: userId }, + billingPeriod: { + start: defaultBillingPeriod().start.toISOString(), + end: defaultBillingPeriod().end.toISOString(), + source: 'default', + }, + payerSubscription: null, + } +} + +function attributedCallback( + attribution: BillingAttributionSnapshot, + requestKey: string, + cost: number +): NextRequest { + return new NextRequest('http://localhost:3000/api/billing/update-cost', { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'x-api-key': 'internal', + 'x-sim-billing-protocol': 'attribution-v1', + 'x-sim-billing-request-id': requestKey, + [BILLING_ATTRIBUTION_HEADER]: serializeBillingAttributionHeader(attribution), + }, + body: JSON.stringify({ + userId: attribution.actorUserId, + workspaceId: attribution.workspaceId, + cost, + model: 'test-model', + source: 'copilot', + idempotencyKey: requestKey, + }), + }) +} + +async function usageExceeded(response: ReturnType): Promise { + const res = await response + expect(res.status).toBe(200) + return (await res.json()).usageExceeded +} + describe('direct-v1 cost callbacks in PostgreSQL', () => { it('records spend after a Stripe rollover in the payer current period', async () => { const now = Date.now() @@ -157,3 +244,38 @@ describe('direct-v1 cost callbacks in PostgreSQL', () => { expect((await requestRows(requestKey)).map((row) => Number(row.cost))).toEqual([0.5]) }) }) + +/** + * Each payer's first callback caches an admission ($1 of a $5 limit), and its second, inside the + * near-limit cache window, records spend that takes it to $6. + */ +describe('usage standing after a charge that crosses the limit', () => { + it('pauses a free direct-v1 run although its admission is cached', async () => { + const payer = await createFreePayer() + + expect(await usageExceeded(POST(callback(payer, generateId(), 1), {}))).toBe(false) + expect(await usageExceeded(POST(callback(payer, generateId(), 5), {}))).toBe(true) + }) + + it('pauses a free attributed run although its admission is cached', async () => { + const attribution = await createFreeWorkspacePayer() + + expect(await usageExceeded(POST(attributedCallback(attribution, generateId(), 1), {}))).toBe( + false + ) + expect(await usageExceeded(POST(attributedCallback(attribution, generateId(), 5), {}))).toBe( + true + ) + }) + + it("keeps serving a paid payer's cached admission", async () => { + const now = Date.now() + const payer = await createPayer({ + start: new Date(now - 10 * DAY_MS), + end: new Date(now + 20 * DAY_MS), + }) + + expect(await usageExceeded(POST(callback(payer, generateId(), 1), {}))).toBe(false) + expect(await usageExceeded(POST(callback(payer, generateId(), 5), {}))).toBe(false) + }) +}) diff --git a/apps/sim/app/api/billing/update-cost/route.test.ts b/apps/sim/app/api/billing/update-cost/route.test.ts index a0098ea6995..4ffd160166d 100644 --- a/apps/sim/app/api/billing/update-cost/route.test.ts +++ b/apps/sim/app/api/billing/update-cost/route.test.ts @@ -931,6 +931,8 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { source: 'stripe' as const, }, } + const PAID_SUBSCRIPTION = { id: 'sub-1', plan: 'team', status: 'active', seats: 4 } + const PAID_ATTRIBUTION = { ...CURRENT_ATTRIBUTION, payerSubscription: PAID_SUBSCRIPTION } function attributedCallback() { callbackSequence += 1 @@ -1005,7 +1007,8 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { }) }) - it('serves a cached admission to every step and re-reads a refusal', async () => { + it('serves a paid payer a cached admission to every step and re-reads a refusal', async () => { + mockRequireBillingAttributionHeader.mockReturnValue(PAID_ATTRIBUTION) mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: false }) expect((await (await POST(attributedCallback())).json()).usageExceeded).toBe(false) @@ -1022,6 +1025,17 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { expect((await (await POST(attributedCallback())).json()).usageExceeded).toBe(false) }) + it('reads a free payer standing past its cached admission after every charge', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: false }) + expect((await (await POST(attributedCallback(), {})).json()).usageExceeded).toBe(false) + + mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: true, scope: 'payer' }) + await expect((await POST(attributedCallback(), {})).json()).resolves.toMatchObject({ + usageExceeded: true, + usageUpgrade: { action: 'upgrade_plan' }, + }) + }) + it('answers a duplicate retry with the verdict its lost first answer carried', async () => { mockRecordCumulativeUsage.mockResolvedValue({ billed: false, delta: 0, total: 0.5 }) mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: true, scope: 'payer' }) @@ -1209,7 +1223,8 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { it.each([ ['an unreadable ledger', { isExceeded: true, reason: 'usage_unavailable' }], ['a blocked account', { isExceeded: true, reason: 'billing_blocked', scope: 'payer' }], - ])('does not pause a run for %s', async (_case, verdict) => { + ])('does not pause a paid run for %s', async (_case, verdict) => { + mockRequireBillingAttributionHeader.mockReturnValue(PAID_ATTRIBUTION) mockCheckAttributedUsageLimits.mockResolvedValue(verdict) const body = await (await POST(attributedCallback())).json() @@ -1218,6 +1233,42 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { expect(body).not.toHaveProperty('usageUpgrade') }) + it.each([ + [ + 'an unreadable ledger', + () => + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + reason: 'usage_unavailable', + }), + ], + [ + 'a failed gate read', + () => mockCheckAttributedUsageLimits.mockRejectedValue(new Error('ledger read timed out')), + ], + ])('pauses a free run for %s, keeping the recorded charge', async (_case, answer) => { + answer() + + const res = await POST(attributedCallback(), {}) + + expect(res.status).toBe(200) + await expect(res.json()).resolves.toMatchObject({ + success: true, + usageExceeded: true, + usageUpgrade: { action: 'upgrade_plan' }, + }) + }) + + it('does not pause a free run for a blocked account', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + reason: 'billing_blocked', + scope: 'payer', + }) + + expect((await (await POST(attributedCallback(), {})).json()).usageExceeded).toBe(false) + }) + it('tells a member over the cap their organization set who can raise it', async () => { mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: true, scope: 'member' }) @@ -1284,8 +1335,11 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { expect(body.usageExceeded).toBe(true) }) - it('keeps a run going when its current period cannot be read', async () => { - mockRequireBillingAttributionHeader.mockReturnValue(STRIPE_ATTRIBUTION) + it('keeps a paid run going when its current period cannot be read', async () => { + mockRequireBillingAttributionHeader.mockReturnValue({ + ...STRIPE_ATTRIBUTION, + payerSubscription: PAID_SUBSCRIPTION, + }) mockRefreshAttributionPeriod.mockRejectedValue(new Error('subscription read timed out')) mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: true, scope: 'payer' }) @@ -1304,31 +1358,39 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { vi.useRealTimers() }) - it('answers not exceeded when the standing read outlasts the callback budget', async () => { - mockCheckAttributedUsageLimits.mockImplementation(async () => { - await sleep(1500) - return { isExceeded: true, scope: 'payer' } - }) + it.each([ + ['a paid', PAID_ATTRIBUTION, false], + ['a free', CURRENT_ATTRIBUTION, true], + ])( + 'answers %s payer whose standing read outlasts the callback budget', + async (_case, attribution, exceeded) => { + mockRequireBillingAttributionHeader.mockReturnValue(attribution) + mockCheckAttributedUsageLimits.mockImplementation(async () => { + await sleep(1500) + return { isExceeded: false } + }) - const pending = POST(attributedCallback()) - await vi.advanceTimersByTimeAsync(1000) - const res = await pending + const pending = POST(attributedCallback(), {}) + await vi.advanceTimersByTimeAsync(1000) + const res = await pending - expect(res.status).toBe(200) - await expect(res.json()).resolves.toMatchObject({ success: true, usageExceeded: false }) - // Settle the abandoned read so no later test coalesces onto it. - await vi.advanceTimersByTimeAsync(500) - }) + expect(res.status).toBe(200) + await expect(res.json()).resolves.toMatchObject({ success: true, usageExceeded: exceeded }) + // Settle the abandoned read so no later test coalesces onto it. + await vi.advanceTimersByTimeAsync(500) + } + ) - it('does not pause a run on a verdict read across the end of its period', async () => { + it('does not pause a paid run on a verdict read across the end of its period', async () => { const straddling = { - ...CURRENT_ATTRIBUTION, + ...PAID_ATTRIBUTION, billingPeriod: { start: '2026-07-01T00:00:00.000Z', end: new Date(Date.now() + 40).toISOString(), source: 'stripe' as const, }, } + mockRequireBillingAttributionHeader.mockReturnValue(PAID_ATTRIBUTION) mockRefreshAttributionPeriod.mockResolvedValue(straddling) mockCheckAttributedUsageLimits.mockImplementation(async () => { await sleep(80) @@ -1363,7 +1425,8 @@ describe('POST /api/billing/update-cost — mid-run usage gate', () => { }) }) - it('keeps a recorded charge successful when the gate read fails', async () => { + it('keeps a paid recorded charge successful and unpaused when the gate read fails', async () => { + mockRequireBillingAttributionHeader.mockReturnValue(PAID_ATTRIBUTION) mockCheckAttributedUsageLimits.mockRejectedValue(new Error('ledger read timed out')) const res = await POST(attributedCallback()) diff --git a/apps/sim/app/api/billing/update-cost/route.ts b/apps/sim/app/api/billing/update-cost/route.ts index ee751b0386e..ce7af14c86a 100644 --- a/apps/sim/app/api/billing/update-cost/route.ts +++ b/apps/sim/app/api/billing/update-cost/route.ts @@ -23,6 +23,7 @@ import { toBillingContext, } from '@/lib/billing/core/attribution' import { + isFreeRunPayer, type MidRunUsageVerdict, readMidRunAccountUsageVerdict, readMidRunUsageVerdict, @@ -56,9 +57,9 @@ import { withIncomingGoSpan } from '@/lib/mothership/request/otel' const logger = createLogger('BillingUpdateCostAPI') /** * How long a cost callback waits on the payer's standing. The worker gives up on the whole - * callback after 15 s, and a cold gate read can wait on the ledger far longer; past this the - * callback answers not-exceeded. The abandoned read keeps running and caches its admission, and - * the next step or re-check reads a refusal again. + * callback after 15 s, and a cold gate read can wait on the ledger far longer; past this a free + * payer's callback answers exceeded and a paid payer's not-exceeded. The abandoned read keeps + * running, and the next step or re-check reads a refusal again. */ const USAGE_STANDING_TIMEOUT_MS = 1000 @@ -84,42 +85,57 @@ function invalidBillingProtocolResponse(requestId: string, span: Span): NextResp * Reads the run payer's standing after a cost callback, so a long run stops at its next step * once it crosses the limit instead of at its next admission, with the card the worker writes * to its log. The payer is the attributed run's, or the one a direct-v1 run was admitted with. - * A duplicate callback answers too: it is often a retry whose first answer was lost. An - * admission is cached per payer and actor (briefly once near the limit) and a refusal is always - * re-read, so most steps cost no ledger read. The charge is - * already recorded when this runs; a gate that cannot answer reports not-exceeded and leaves the - * refusal to the next step or re-check rather than ending a paying run on a database blip, - * and so does a verdict read that outlasts {@link USAGE_STANDING_TIMEOUT_MS}. An exceeded - * verdict always pauses the run. + * A duplicate callback answers too: it is often a retry whose first answer was lost. The charge + * is already recorded when this runs. + * + * A free payer's spend past its limit is never billed, so its standing is read from the ledger + * on every callback, and a standing that cannot be read (a gate that cannot answer, or a read + * that outlasts {@link USAGE_STANDING_TIMEOUT_MS}) pauses the run. A paid payer's admission is + * cached per payer and actor (briefly once near the limit) and a refusal is always re-read, so + * most steps cost no ledger read; a standing that cannot be read answers not-exceeded and leaves + * the refusal to the next step or re-check rather than ending a paying run on a database blip. + * An exceeded verdict always pauses the run. */ async function readUsageStanding( userId: string, billingAttribution: BillingAttributionSnapshot | undefined, accountDecision: AccountBillingDecision | undefined ): Promise { + const readOptions = { uncachedForFreePayer: true } + const payer = billingAttribution ?? accountDecision const readVerdict = billingAttribution - ? () => readMidRunUsageVerdict(billingAttribution) + ? () => readMidRunUsageVerdict(billingAttribution, 0, readOptions) : accountDecision - ? () => readMidRunAccountUsageVerdict(accountDecision) + ? () => readMidRunAccountUsageVerdict(accountDecision, readOptions) : null - if (!isHosted || !readVerdict) return { usageExceeded: false } + if (!isHosted || !payer || !readVerdict) return { usageExceeded: false } + const freePayer = isFreeRunPayer(payer) let verdict: MidRunUsageVerdict try { verdict = await withinDeadline(readVerdict, Date.now() + USAGE_STANDING_TIMEOUT_MS) } catch { - logger.warn('Usage standing read outlasted the callback budget; answering not exceeded') - return { usageExceeded: false } + logger.warn('Usage standing read outlasted the callback budget', { freePayer }) + verdict = { status: 'unknown' } } - // Only a spent limit pauses the run. A blocked account is refused at the run's next - // continuation or re-check, with blocked-account copy rather than the upgrade card. - if (verdict.status !== 'exceeded') return { usageExceeded: false } + if (verdict.status === 'exceeded') { + return { + usageExceeded: true, + usageUpgrade: await resolveUsageUpgradePayload( + userId, + billingAttribution ?? verdict.payer, + verdict.scope + ), + } + } + // A blocked account is refused at the run's next continuation or re-check, with + // blocked-account copy rather than the upgrade card. + if (verdict.status !== 'unknown' || !freePayer) return { usageExceeded: false } return { usageExceeded: true, - usageUpgrade: await resolveUsageUpgradePayload( - userId, - billingAttribution ?? verdict.payer, - verdict.scope - ), + usageUpgrade: await resolveUsageUpgradePayload(userId, { + billingEntity: payer.billingEntity, + payerSubscription: null, + }), } } diff --git a/apps/sim/app/api/copilot/api-keys/validate/route.test.ts b/apps/sim/app/api/copilot/api-keys/validate/route.test.ts index 04c6a3b2b4f..1a5f8d643be 100644 --- a/apps/sim/app/api/copilot/api-keys/validate/route.test.ts +++ b/apps/sim/app/api/copilot/api-keys/validate/route.test.ts @@ -652,9 +652,21 @@ describe('validation lifecycle purposes', () => { } }) - it('admits a direct-v1 continuation whose usage cannot be read', async () => { + it('admits a paid direct-v1 continuation whose usage cannot be read', async () => { mockCheckUsageStatus.mockResolvedValue({ isExceeded: true, unavailable: true }) - expect((await POST(request(body, directHeaders))).status).toBe(200) + const paidHeaders = { + ...directHeaders, + 'x-sim-billing-account-decision': encode({ + ...ACCOUNT_BILLING_DECISION, + payerSubscriptionId: 'sub-account-org', + }), + } + expect((await POST(request(body, paidHeaders), {})).status).toBe(200) + }) + + it('refuses a free direct-v1 continuation whose usage cannot be read', async () => { + mockCheckUsageStatus.mockResolvedValue({ isExceeded: true, unavailable: true }) + expect((await POST(request(body, directHeaders), {})).status).toBe(402) }) it.each([ @@ -771,14 +783,44 @@ describe('validation lifecycle purposes', () => { }) }) - it('admits a polled continuation whose spend cannot be read', async () => { + it('admits a paid payer polled continuation whose spend cannot be read', async () => { + const paidHeaders = { + ...attributedHeaders, + 'x-sim-billing-attribution': encode({ + ...ATTRIBUTION, + payerSubscription: { + id: 'sub-org-1', + referenceId: 'org-1', + plan: 'team', + status: 'active', + seats: 5, + periodStart: null, + periodEnd: null, + }, + }), + } mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: true, reason: 'usage_unavailable', }) - expect((await POST(request(body, attributedHeaders))).status).toBe(200) + expect((await POST(request(body, paidHeaders), {})).status).toBe(200) mockCheckAttributedUsageLimits.mockRejectedValue(new Error('ledger read timed out')) - expect((await POST(request(body, attributedHeaders))).status).toBe(200) + expect((await POST(request(body, paidHeaders), {})).status).toBe(200) + }) + + it('refuses a free payer polled continuation whose spend cannot be read, with the usage card', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + reason: 'usage_unavailable', + }) + const unavailable = await POST(request(body, attributedHeaders), {}) + expect(unavailable.status).toBe(402) + await expect(unavailable.json()).resolves.toMatchObject({ + code: 'USAGE_LIMIT_EXCEEDED', + usageUpgrade: { action: 'upgrade_plan' }, + }) + mockCheckAttributedUsageLimits.mockRejectedValue(new Error('ledger read timed out')) + expect((await POST(request(body, attributedHeaders), {})).status).toBe(402) }) it('refuses a continuation over its usage limit with the card the worker writes', async () => { diff --git a/apps/sim/app/api/copilot/api-keys/validate/route.ts b/apps/sim/app/api/copilot/api-keys/validate/route.ts index ad9509666f8..e52ba5e5dde 100644 --- a/apps/sim/app/api/copilot/api-keys/validate/route.ts +++ b/apps/sim/app/api/copilot/api-keys/validate/route.ts @@ -27,6 +27,7 @@ import { serializeBillingAttributionHeader, } from '@/lib/billing/core/attribution' import { + isFreeRunPayer, readMidRunAccountUsageVerdict, readMidRunUsageVerdict, } from '@/lib/billing/core/mid-run-usage' @@ -418,7 +419,8 @@ export const POST = withRouteHandler((req: NextRequest) => }) // A continuation, and a worker's periodic re-check of a long run, also reads the // original payer's spend through the cached execution usage gate. A read that fails - // admits: the run is already under way, and the next re-check reads again. + // refuses a free payer, whose spend past the limit is never billed, and admits a paid + // one: the run is already under way, and the next re-check reads again. const verdict = !blocked?.blocked && purpose === COPILOT_VALIDATION_PURPOSE.continuation && billing ? billing.kind === 'attributed' @@ -444,8 +446,15 @@ export const POST = withRouteHandler((req: NextRequest) => { status: 402 } ) } - if (verdict?.status === 'exceeded') { - logger.info('[API VALIDATION] Continuation usage exceeded', { userId }) + const unreadFreePayer = + verdict?.status === 'unknown' && + billing !== null && + isFreeRunPayer(billing.kind === 'attributed' ? billing.attribution : billing.decision) + if (verdict?.status === 'exceeded' || unreadFreePayer) { + logger.info('[API VALIDATION] Continuation usage exceeded', { + userId, + unread: unreadFreePayer, + }) span.setAttribute( TraceAttr.CopilotValidateOutcome, CopilotValidateOutcome.UsageExceeded @@ -453,8 +462,14 @@ export const POST = withRouteHandler((req: NextRequest) => span.setAttribute(TraceAttr.HttpStatusCode, 402) const usageUpgrade = await resolveUsageUpgradePayload( userId, - billing?.kind === 'attributed' ? billing.attribution : verdict.payer, - verdict.scope + billing?.kind === 'attributed' + ? billing.attribution + : verdict?.status === 'exceeded' + ? verdict.payer + : billing + ? { billingEntity: billing.decision.billingEntity, payerSubscription: null } + : undefined, + verdict?.status === 'exceeded' ? verdict.scope : undefined ) return NextResponse.json( { diff --git a/apps/sim/lib/billing/concurrency-defaults.ts b/apps/sim/lib/billing/concurrency-defaults.ts index 35a21c7c318..165812aa986 100644 --- a/apps/sim/lib/billing/concurrency-defaults.ts +++ b/apps/sim/lib/billing/concurrency-defaults.ts @@ -6,7 +6,7 @@ import type { PlanCategory } from '@/lib/billing/plan-helpers' * and `team` is the Max tier covering Max and Max for Teams. */ export const DEFAULT_BILLING_CONCURRENCY_LIMITS = { - free: 10, + free: 2, pro: 50, team: 200, enterprise: 1000, diff --git a/apps/sim/lib/billing/core/mid-run-usage.ts b/apps/sim/lib/billing/core/mid-run-usage.ts index f13d96ab97d..c89748f3267 100644 --- a/apps/sim/lib/billing/core/mid-run-usage.ts +++ b/apps/sim/lib/billing/core/mid-run-usage.ts @@ -7,6 +7,7 @@ import { type AttributedUsageLimitsResult, type BillingAttributionSnapshot, checkAccountBillingBlocks, + checkAttributedUsageLimits, refreshAttributionPeriod, } from '@/lib/billing/core/attribution' import { getOrganizationSubscription } from '@/lib/billing/core/billing' @@ -19,9 +20,11 @@ import { USAGE_GATE_TTL_MS, usageAdmissionTtlMs, } from '@/lib/billing/core/usage-gate-cache' +import { isFree } from '@/lib/billing/plan-helpers' import type { UsageUpgradePayer } from '@/lib/billing/usage-upgrade' import { coalesceLocally } from '@/lib/concurrency/singleflight' import { isBillingEnabled, isHosted } from '@/lib/core/config/env-flags' +import { withinDeadline } from '@/lib/core/utils/deadline' const logger = createLogger('MidRunUsage') @@ -33,8 +36,9 @@ const logger = createLogger('MidRunUsage') * - `blocked`: the account is blocked (payment failed, dispute); the run is refused as a blocked * account, never with the upgrade card. * - `unknown`: usage, or the payer's current period, could not be read. Admission fails closed - * on an unreadable ledger, but a run already under way continues: a database blip must not - * end a paying user's long run, and the next step or re-check reads again. + * on an unreadable ledger. Callers stop a free payer's run on it (see {@link isFreeRunPayer}), + * whose spend past the limit is never billed, and let a paid payer's run continue: a database + * blip must not end a paying user's long run, and the next step or re-check reads again. */ export type MidRunUsageVerdict = | { status: 'within' } @@ -46,6 +50,28 @@ export type MidRunUsageVerdict = | { status: 'blocked'; message?: string } | { status: 'unknown' } +interface MidRunUsageReadOptions { + /** + * Reads a free payer's standing from the ledger rather than a cached admission, for a caller + * that has just recorded a charge: a free payer is always near its limit, and a cached + * admission would let that charge run past it unseen. Paid payers keep the cached admission. + */ + uncachedForFreePayer?: boolean +} + +/** + * Whether a run's payer is on the free plan, as admitted. An attributed run reads its payer + * subscription (an organization payer without one is free). A direct-v1 decision records a + * subscription only for a payer with an entitled one, and a free payer has none. + */ +export function isFreeRunPayer( + payer: BillingAttributionSnapshot | AccountBillingDecision +): boolean { + return 'payerSubscription' in payer + ? isFree(payer.payerSubscription?.plan) + : !payer.payerSubscriptionId +} + /** * How long a payer's current period stays cached for mid-run checks. Every model step settles a * cost callback that reads it; a subscription period change (rollover, anchor reset) reaches the @@ -112,13 +138,20 @@ function limitReachedWith( async function readGateVerdict( attribution: BillingAttributionSnapshot, - unrecordedCost: number + unrecordedCost: number, + options: MidRunUsageReadOptions ): Promise { let usage: AttributedUsageLimitsResult try { - usage = await checkExecutionUsageLimits(attribution) + usage = + options.uncachedForFreePayer && isFreeRunPayer(attribution) + ? await withinDeadline( + () => checkAttributedUsageLimits(attribution), + Date.now() + USAGE_GATE_SETTLE_TIMEOUT_MS + ) + : await checkExecutionUsageLimits(attribution) } catch (error) { - logger.warn('Mid-run usage read failed; continuing the run', { + logger.warn('Mid-run usage read failed', { error: getErrorMessage(error), }) return { status: 'unknown' } @@ -131,7 +164,7 @@ async function readGateVerdict( return { status: 'blocked', ...(usage.message ? { message: usage.message } : {}) } } if (usage.reason === 'usage_unavailable') { - logger.warn('Mid-run usage could not be read; continuing the run') + logger.warn('Mid-run usage could not be read') return { status: 'unknown' } } return { status: 'exceeded', ...(usage.scope ? { scope: usage.scope } : {}) } @@ -141,8 +174,8 @@ async function readGateVerdict( * Judges a run against the period its charges land in. A Stripe-period payer's charges roll into * whatever period the subscription is in now (a rollover or an early anchor reset included), so * such a run is judged against the payer's CURRENT period. A current period that cannot be read, - * or that ends before its verdict is read, makes the verdict unknown, so the run continues and the - * next callback judges the next period. Any other payer's charges stay in the admitted + * or that ends before its verdict is read, makes the verdict unknown, and the next callback judges + * the next period. Any other payer's charges stay in the admitted * period (a reporting window, or the open default one), so that period is judged, even after it * ends. * @@ -150,23 +183,24 @@ async function readGateVerdict( */ export async function readMidRunUsageVerdict( attribution: BillingAttributionSnapshot, - unrecordedCost = 0 + unrecordedCost = 0, + options: MidRunUsageReadOptions = {} ): Promise { if (!isHosted || !isBillingEnabled) return { status: 'within' } if (!rollsIntoCurrentPeriod(attribution.billingPeriod)) { - return readGateVerdict(attribution, unrecordedCost) + return readGateVerdict(attribution, unrecordedCost, options) } let judged: BillingAttributionSnapshot try { judged = await currentAttribution(attribution) } catch (error) { - logger.warn('Current billing period could not be read; continuing the run', { + logger.warn('Current billing period could not be read', { error: getErrorMessage(error), }) return { status: 'unknown' } } if (periodHasEnded(judged)) return { status: 'unknown' } - const verdict = await readGateVerdict(judged, unrecordedCost) + const verdict = await readGateVerdict(judged, unrecordedCost, options) // A period that ended during the read is judged at the next callback, which reloads it. return periodHasEnded(judged) ? { status: 'unknown' } : verdict } @@ -187,9 +221,12 @@ const accountVerdictCache = new LRUCache({ * payer, in the gate's order: a blocked actor or payer first, then the payer's spend. The payer * is the one saved in the decision at admission, never re-selected from the actor's current * memberships, and the period judged is the one its charges land in, as for attributed runs. + * Whether {@link MidRunUsageReadOptions.uncachedForFreePayer} applies is judged by the plan the + * run was admitted with, as {@link isFreeRunPayer} judges it. */ export async function readMidRunAccountUsageVerdict( - decision: AccountBillingDecision + decision: AccountBillingDecision, + options: MidRunUsageReadOptions = {} ): Promise { if (!isHosted || !isBillingEnabled) return { status: 'within' } try { @@ -220,7 +257,8 @@ export async function readMidRunAccountUsageVerdict( subscription?.status ?? '', subscription?.seats ?? '', ].join(':') - const cached = accountVerdictCache.get(key) + const uncached = options.uncachedForFreePayer === true && isFreeRunPayer(decision) + const cached = uncached ? undefined : accountVerdictCache.get(key) if (cached) return cached const block = await checkAccountBillingBlocks(decision) if (block.blocked) { @@ -240,15 +278,16 @@ export async function readMidRunAccountUsageVerdict( periodEnd: billingPeriod.end, } : null) - const usage = await coalesceLocally( - `mid-run-account-usage:${key}`, - () => - checkUsageStatus(decision.userId, usageSubscription, { - billingEntity: payer, - billingPeriod, - }), - USAGE_GATE_SETTLE_TIMEOUT_MS - ) + const readUsage = () => + checkUsageStatus(decision.userId, usageSubscription, { billingEntity: payer, billingPeriod }) + // A shared in-flight read may have started before this caller's charge was recorded. + const usage = uncached + ? await withinDeadline(readUsage, Date.now() + USAGE_GATE_SETTLE_TIMEOUT_MS) + : await coalesceLocally( + `mid-run-account-usage:${key}`, + readUsage, + USAGE_GATE_SETTLE_TIMEOUT_MS + ) if (usage.unavailable) return { status: 'unknown' } if (usage.isExceeded) { return { @@ -261,7 +300,7 @@ export async function readMidRunAccountUsageVerdict( accountVerdictCache.set(key, within, { ttl: usageAdmissionTtlMs(usage) }) return within } catch (error) { - logger.warn('Mid-run account usage read failed; continuing the run', { + logger.warn('Mid-run account usage read failed', { error: getErrorMessage(error), }) return { status: 'unknown' } diff --git a/apps/sim/lib/compare/data/sim.ts b/apps/sim/lib/compare/data/sim.ts index a25d488304d..7d904bff03b 100644 --- a/apps/sim/lib/compare/data/sim.ts +++ b/apps/sim/lib/compare/data/sim.ts @@ -1152,7 +1152,7 @@ export const simProfile: CompetitorProfile = { }, executionLimits: { value: - 'Plan-gated: synchronous runs time out at 5 minutes on the free plan and 50 minutes on paid plans; async runs default to 90 minutes, while Enterprise policy can be configured up to 7 days; concurrency ranges from 10 to 1,000 executions per billing account depending on plan', + 'Plan-gated: synchronous runs time out at 5 minutes on the free plan and 50 minutes on paid plans; async runs default to 90 minutes, while Enterprise policy can be configured up to 7 days; concurrency ranges from 2 to 1,000 executions per billing account depending on plan', detail: 'A direct async API request can shorten the account policy for that request, but cannot extend it. Concurrency limits are published in the platform cost docs, where the Enterprise limit is 1,000 by default and customizable. A run that exceeds its time limit is terminated and marked failed with a timeout error.', shortValue: '5-50 min sync; async defaults to 90 min, Enterprise up to 7 days', diff --git a/apps/sim/lib/mothership/request/lifecycle/admission.test.ts b/apps/sim/lib/mothership/request/lifecycle/admission.test.ts index 53ea16e46ab..2f0df88200b 100644 --- a/apps/sim/lib/mothership/request/lifecycle/admission.test.ts +++ b/apps/sim/lib/mothership/request/lifecycle/admission.test.ts @@ -112,15 +112,40 @@ describe('continuation admission', () => { await authorizeLifecycleContinuation(context) expect(mockCheckExecutionUsageLimits).not.toHaveBeenCalled() }) - it('continues a leg when spend cannot be read', async () => { + it('continues a paid payer leg when spend cannot be read', async () => { + const paid = { + ...context, + billingAttribution: { + ...attribution, + payerSubscription: { + id: 'sub-org', + referenceId: 'original-org', + plan: 'team', + status: 'active', + seats: 4, + periodStart: attribution.billingPeriod.start, + periodEnd: attribution.billingPeriod.end, + }, + }, + } mockCheckExecutionUsageLimits.mockRejectedValueOnce(new Error('ledger read timed out')) - await expect(authorizeLifecycleContinuation(context)).resolves.toBeUndefined() + await expect(authorizeLifecycleContinuation(paid)).resolves.toBeUndefined() mockCheckExecutionUsageLimits.mockResolvedValueOnce({ isExceeded: true, reason: 'usage_unavailable', }) - await expect(authorizeLifecycleContinuation(context)).resolves.toBeUndefined() + await expect(authorizeLifecycleContinuation(paid)).resolves.toBeUndefined() + }) + it('refuses a free payer leg when spend cannot be read', async () => { + mockCheckExecutionUsageLimits.mockRejectedValueOnce(new Error('ledger read timed out')) + await expect(authorizeLifecycleContinuation(context)).rejects.toBeInstanceOf(BillingLimitError) + + mockCheckExecutionUsageLimits.mockResolvedValueOnce({ + isExceeded: true, + reason: 'usage_unavailable', + }) + await expect(authorizeLifecycleContinuation(context)).rejects.toBeInstanceOf(BillingLimitError) }) it('refuses a payer the gate finds blocked as a blocked account, not with the usage card', async () => { mockCheckExecutionUsageLimits.mockResolvedValueOnce({ @@ -142,6 +167,15 @@ describe('continuation admission', () => { end: '2026-08-01T00:00:00.000Z', source: 'stripe' as const, }, + payerSubscription: { + id: 'sub-org', + referenceId: 'original-org', + plan: 'team', + status: 'active', + seats: 4, + periodStart: '2026-07-01T00:00:00.000Z', + periodEnd: '2026-08-01T00:00:00.000Z', + }, } mockGetOrganizationSubscription.mockResolvedValue({ id: 'sub-org', diff --git a/apps/sim/lib/mothership/request/lifecycle/admission.ts b/apps/sim/lib/mothership/request/lifecycle/admission.ts index 13301cc36c2..69ea8839441 100644 --- a/apps/sim/lib/mothership/request/lifecycle/admission.ts +++ b/apps/sim/lib/mothership/request/lifecycle/admission.ts @@ -6,7 +6,7 @@ import { COPILOT_BILLING_PROTOCOL_HEADER, requireBillingCallbackAttribution, } from '@/lib/billing/core/attribution' -import { readMidRunUsageVerdict } from '@/lib/billing/core/mid-run-usage' +import { isFreeRunPayer, readMidRunUsageVerdict } from '@/lib/billing/core/mid-run-usage' import { isHosted } from '@/lib/core/config/env-flags' import { OrchestrationError } from '@/lib/core/orchestration/types' import { @@ -84,5 +84,9 @@ export async function authorizeLifecycleContinuation( if (usage.status === 'blocked') throw new OrchestrationError('forbidden', 'Continuation billing account is blocked') if (usage.status === 'exceeded') throw new BillingLimitError(context.userId, usage.scope) + // A free payer's spend past its limit is never billed, so an unreadable standing refuses it. + if (usage.status === 'unknown' && isFreeRunPayer(context.billingAttribution)) { + throw new BillingLimitError(context.userId) + } } } diff --git a/apps/sim/providers/run-usage-limit.test.ts b/apps/sim/providers/run-usage-limit.test.ts index 3fd7e5f2a29..81b3083254f 100644 --- a/apps/sim/providers/run-usage-limit.test.ts +++ b/apps/sim/providers/run-usage-limit.test.ts @@ -2,6 +2,10 @@ * @vitest-environment node */ import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing' +import { + billingAttributionMock, + billingAttributionMockFns, +} from '@sim/testing/mocks/billing-attribution.mock' import { billingCoreMock } from '@sim/testing/mocks/billing-core.mock' import { billingUsageGateCacheMock, @@ -17,6 +21,7 @@ vi.mock('@/providers/registry', () => ({ getProviderExecutor: async () => ({ executeRequest: mockExecuteRequest }), })) vi.mock('@/lib/billing/core/usage-gate-cache', () => billingUsageGateCacheMock) +vi.mock('@/lib/billing/core/attribution', () => billingAttributionMock) vi.mock('@/tools', () => toolsMock) vi.mock('@/lib/billing/core/billing', () => billingCoreMock) vi.mock('@/providers/cost-policy', async (importOriginal) => ({ @@ -60,6 +65,7 @@ import { providerFailureLogLevel } from '@/providers/streaming-tool-loop-shared' import { ProviderError, type ProviderRequest, type ProviderToolConfig } from '@/providers/types' const { mockCheckExecutionUsageLimits } = billingUsageGateCacheMockFns +const { mockCheckAttributedUsageLimits } = billingAttributionMockFns const { mockExecuteTool } = toolsMockFns const attribution: BillingAttributionSnapshot = { @@ -105,6 +111,11 @@ async function modelStep(request: ProviderRequest, input: number, output: number beforeEach(() => { setEnvFlags({ isHosted: true, isBillingEnabled: true }) mockCheckExecutionUsageLimits.mockReset() + // A free payer's mid-run read skips the cached gate; both answer from the same ledger. + mockCheckAttributedUsageLimits.mockReset() + mockCheckAttributedUsageLimits.mockImplementation((attribution) => + mockCheckExecutionUsageLimits(attribution) + ) mockCheckExecutionUsageLimits.mockResolvedValue({ isExceeded: false, payerUsage: { currentUsage: 3, limit: 5 }, @@ -436,22 +447,30 @@ describe('run usage limit', () => { reason: 'usage_unavailable', }), ], - [ - 'the account is blocked', - () => - mockCheckExecutionUsageLimits.mockResolvedValueOnce({ - isExceeded: true, - reason: 'billing_blocked', - scope: 'payer', - }), - ], - ])('keeps the run going when %s, without reading again for a while', async (_case, answer) => { + ])( + 'stops the free run when %s, since its spend past the limit is never billed', + async (_case, answer) => { + const request = startRequest(startRun()) + await modelStep(request, 0.1, 0.1) + answer() + + await expect( + prepareConversationGeneration(request, 'anthropic', { messages: [] }) + ).rejects.toThrow('usage limit') + } + ) + + it('keeps the run going when the account is blocked, without reading again for a while', async () => { vi.useFakeTimers({ toFake: ['Date'] }) try { const request = startRequest(startRun()) await modelStep(request, 1, 1) mockCheckExecutionUsageLimits.mockClear() - answer() + mockCheckExecutionUsageLimits.mockResolvedValueOnce({ + isExceeded: true, + reason: 'billing_blocked', + scope: 'payer', + }) await expect(modelStep(request, 1, 1)).resolves.toBeUndefined() await expect(modelStep(request, 1, 1)).resolves.toBeUndefined() diff --git a/apps/sim/providers/run-usage-limit.ts b/apps/sim/providers/run-usage-limit.ts index b971248d29c..cb8434e886d 100644 --- a/apps/sim/providers/run-usage-limit.ts +++ b/apps/sim/providers/run-usage-limit.ts @@ -67,9 +67,10 @@ export function recordModelStepCost( * and a free plan's spend past its limit is never billed. A paid payer's overshoot is billed as * overage, so its runs are left to admission and never read here. * - * Reads nothing until the run has spent, then the cached execution gate. A gate that cannot - * answer, or an account that is blocked, lets the run continue and is not read again for - * {@link NEAR_LIMIT_USAGE_GATE_TTL_MS}: the next admission refuses it. The stop is not retried, + * Reads nothing until the run has spent, then the payer's ledger. A gate that cannot + * answer stops the run as if the limit were spent. A blocked account lets the run continue and + * is not read again for {@link NEAR_LIMIT_USAGE_GATE_TTL_MS}: the next admission refuses it with + * blocked-account copy rather than the upgrade card. The stop is not retried, * it carries the cost of the request's settled steps and tool calls so the stopped block is * billed for them, and it is attributed to the user so it logs at info, not error. */ @@ -91,18 +92,21 @@ export async function assertRunWithinUsageLimit( return } - const verdict = await readMidRunUsageVerdict(attribution, unrecorded.total) + const verdict = await readMidRunUsageVerdict(attribution, unrecorded.total, { + uncachedForFreePayer: true, + }) // A run cancelled during the read keeps its cancellation; the provider call then aborts. if (verdict.status === 'within' || executionContext.abortSignal?.aborted) return - if (verdict.status !== 'exceeded') { + if (verdict.status === 'blocked') { unrecorded.readDeferredUntil = Date.now() + NEAR_LIMIT_USAGE_GATE_TTL_MS return } + const exceeded = verdict.status === 'exceeded' ? verdict : undefined const { message } = await resolveUsageUpgradePayload( attribution.actorUserId, - verdict.payer ?? attribution, - verdict.scope + exceeded?.payer ?? attribution, + exceeded?.scope ) const error = markFailureKind(new RunUsageLimitError(message), 'user') if (runtime.settledCost?.total) attachTrustedExecutionCost(error, runtime.settledCost) diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json index a28f6195bb1..23ed185abe0 100644 --- a/scripts/check-test-types.baseline.json +++ b/scripts/check-test-types.baseline.json @@ -71,7 +71,7 @@ "TS2554": 1 }, "apps/sim/app/api/billing/update-cost/route.test.ts": { - "TS2554": 51 + "TS2554": 50 }, "apps/sim/app/api/billing/update-cost/worker.live.test.ts": { "TS2554": 1 @@ -98,7 +98,7 @@ "TS2554": 1 }, "apps/sim/app/api/copilot/api-keys/validate/route.test.ts": { - "TS2554": 55 + "TS2554": 52 }, "apps/sim/app/api/copilot/chat/stop/route.test.ts": { "TS18046": 2,