fix(billing): stop free-plan workflow runs at their usage limit mid-run

apps/sim/executor/execution/block-executor.ts

@@ -3,7 +3,12 @@ import { describeError, toError } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { isRecordLike, toRecord } from '@sim/utils/object'
 import { findDatabaseQueryError } from '@/lib/core/errors/database-query-error'
-import { logFailureOnce, markFailureKind, markFailureLogged } from '@/lib/core/errors/failure-log'
+import {
+  classifyFailure,
+  logFailureOnce,
+  markFailureKind,
+  markFailureLogged,
+} from '@/lib/core/errors/failure-log'
 import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types'
 import { redactApiKeys } from '@/lib/core/security/redaction'
 import { normalizeStringArray } from '@/lib/core/utils/arrays'
@@ -1327,10 +1332,13 @@ export class BlockExecutor {
     try {
       pumpResult = await pump.run()
     } catch (error) {
-      this.execLogger.error('Error reading stream for block', {
-        blockId,
-        ...projectStreamDiagnosticError(error),
-      })
+      this.execLogger[classifyFailure(error) === 'user' ? 'info' : 'error'](
+        'Error reading stream for block',
+        {
+          blockId,
+          ...projectStreamDiagnosticError(error),
+        }
+      )
       if (onStreamPromise) {
         await onStreamPromise.catch(() => {})
       }

apps/sim/executor/execution/executor.ts

@@ -485,6 +485,7 @@ export class DAGExecutor {
       executionOrderCounter: {
         value: restoredBlockLogs.reduce((max, log) => Math.max(max, log.executionOrder ?? 0), 0),
       },
+      unrecordedModelCost: this.contextExtensions.unrecordedModelCost ?? { total: 0 },
       metadata: {
         ...this.contextExtensions.metadata,
         ...(this.contextExtensions.billingAttribution

apps/sim/executor/execution/types.ts

@@ -11,6 +11,7 @@ import type {
   NormalizedBlockOutput,
   StartBlockRunMetadata,
   StreamingExecution,
+  UnrecordedModelCost,
 } from '@/executor/types'
 import type {
   ResolvedSecretTraceProvenanceV1,
@@ -329,6 +330,8 @@ export interface ContextExtensions {
    * Takes precedence over `metadata.billingAttribution` when both are set.
    */
   billingAttribution?: BillingAttributionSnapshot
+  /** The invoking run's {@link ExecutionContext.unrecordedModelCost}, for a child billed on its log. */
+  unrecordedModelCost?: UnrecordedModelCost
   stream?: boolean
   selectedOutputs?: string[]
   edges?: Array<{ source: string; target: string }>

apps/sim/executor/handlers/workflow/custom-block-tool-runner.ts

@@ -6,7 +6,11 @@ import { isPlainRecord, isRecordLike } from '@sim/utils/object'
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/attribution'
 import type { PiiBlockOutputRedaction } from '@/executor/execution/types'
 import { WorkflowBlockHandler } from '@/executor/handlers/workflow/handler'
-import type { ExecutionContext, ExecutorDelegationOrigin } from '@/executor/types'
+import type {
+  ExecutionContext,
+  ExecutorDelegationOrigin,
+  UnrecordedModelCost,
+} from '@/executor/types'
 import { projectResolvedSecretDiagnosticContent } from '@/executor/utils/resolved-secret-content-projection'
 import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 import type { SerializedBlock } from '@/serializer/types'
@@ -76,6 +80,8 @@ export function buildCustomBlockExecutionContext(
     principal?: WorkflowExecutionPrincipal
     /** The invoking run's in-flight block-output redaction policy. */
     piiBlockOutputRedaction?: PiiBlockOutputRedaction
+    /** The invoking run's unrecorded model spend, which a child billed on its log adds to. */
+    unrecordedModelCost?: UnrecordedModelCost
   }
 ): ExecutionContext {
   // Prefer the invoking agent run's ids so correlation and cancellation both
@@ -98,6 +104,7 @@ export function buildCustomBlockExecutionContext(
     resolvedSecretTraceRegistry: options.resolvedSecretTraceRegistry,
     environmentVariables: options.environmentVariables,
     piiBlockOutputRedaction: options.piiBlockOutputRedaction,
+    unrecordedModelCost: options.unrecordedModelCost,
     blockStates: new Map(),
     executedBlocks: new Set(),
     blockLogs: [],

apps/sim/executor/handlers/workflow/handler.test.ts

@@ -398,6 +398,7 @@ describe('WorkflowBlockHandler', () => {
         workspaceId: 'workspace-parent',
         environmentVariables: { MY_API_KEY: 'parent-secret' },
         piiBlockOutputRedaction,
+        unrecordedModelCost: { total: 1 },
       } as unknown as ExecutionContext
 
       mockFetch.mockResolvedValueOnce({
@@ -425,6 +426,7 @@ describe('WorkflowBlockHandler', () => {
       expect(executorOptions[0].contextExtensions.piiBlockOutputRedaction).toBe(
         piiBlockOutputRedaction
       )
+      expect(executorOptions[0].contextExtensions.unrecordedModelCost).toBe(ctx.unrecordedModelCost)
       expect(mockGetPersonalAndWorkspaceEnv).not.toHaveBeenCalled()
     })
 
@@ -439,6 +441,7 @@ describe('WorkflowBlockHandler', () => {
         ...mockContext,
         workspaceId: 'workspace-consumer',
         metadata: { ...mockContext.metadata, billingAttribution: consumerAttribution },
+        unrecordedModelCost: { total: 1 },
       } as unknown as ExecutionContext
 
       mockGetCustomBlockAuthority.mockResolvedValue({
@@ -506,6 +509,7 @@ describe('WorkflowBlockHandler', () => {
       })
       expect(executorOptions).toHaveLength(1)
       expect(executorOptions[0].contextExtensions.billingAttribution).toBe(sourceAttribution)
+      expect(executorOptions[0].contextExtensions.unrecordedModelCost).toBeUndefined()
       expect(executorOptions[0].contextExtensions.userId).toBe('owner-9')
       expect(executorOptions[0].contextExtensions.workspaceId).toBe('workspace-source')
       expect(executorOptions[0].contextExtensions.executorDelegationOrigin).toEqual({

apps/sim/executor/handlers/workflow/handler.ts

@@ -886,6 +886,9 @@ export class WorkflowBlockHandler implements BlockHandler {
           // internal tool calls (knowledge, guardrails, MCP, Mothership) can
           // attach the required billing attribution header.
           billingAttribution: childBillingAttribution,
+          // A custom block's child writes its own log and ledger rows; any other child's cost
+          // lands on this run's log, so it spends against this run's unrecorded total.
+          unrecordedModelCost: childExecutionId ? undefined : ctx.unrecordedModelCost,
           resolvedSecretTraceRegistry: childResolvedSecretTraceRegistry,
           // Fall back to the inherited metadata so a toggle-off intermediate
           // child still carries the trusted identity chain to deeper children.

apps/sim/executor/handlers/workflow/tool-runner.test.ts

@@ -70,6 +70,18 @@ describe('runWorkflowTool execution context', () => {
     expect(ctxArg.principal).toEqual(trustedPrincipal)
     expect(ctxArg.metadata.principal).toEqual(trustedPrincipal)
   })
+
+  it("hands the child the invoking run's unrecorded model total itself, not a copy", async () => {
+    const unrecordedModelCost = { total: 1 }
+
+    await runWorkflowTool(
+      { workflowId: 'wf-child', _context: { workspaceId: 'ws-1' } },
+      { environmentVariables: {}, unrecordedModelCost }
+    )
+
+    const [ctxArg] = mockExecute.mock.calls[0]
+    expect(ctxArg.unrecordedModelCost).toBe(unrecordedModelCost)
+  })
 })
 
 const CHILD_SECRET = 'sk-live-7f2c9a41d8be4055b6c3'

apps/sim/executor/handlers/workflow/tool-runner.ts

@@ -12,7 +12,7 @@ import {
   type CustomBlockExecutorContext,
 } from '@/executor/handlers/workflow/custom-block-tool-runner'
 import { WorkflowBlockHandler } from '@/executor/handlers/workflow/handler'
-import type { ExecutorDelegationOrigin } from '@/executor/types'
+import type { ExecutorDelegationOrigin, UnrecordedModelCost } from '@/executor/types'
 import { classifyExecutionError } from '@/executor/utils/errors'
 import { parseJSON } from '@/executor/utils/json'
 import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
@@ -99,6 +99,7 @@ export async function runWorkflowTool(
     executorDelegationOrigin?: ExecutorDelegationOrigin
     principal?: WorkflowExecutionPrincipal
     piiBlockOutputRedaction?: PiiBlockOutputRedaction
+    unrecordedModelCost?: UnrecordedModelCost
   }
 ): Promise<ToolResponse> {
   if (!params.workflowId) {

apps/sim/executor/types.ts

@@ -668,6 +668,20 @@ export interface ExecutionContext {
    * Starts at 0 and increments for each block. Use getNextExecutionOrder() to access.
    */
   executionOrderCounter?: { value: number }
+
+  /**
+   * Billed model spend not yet in the usage ledger, which receives a run's cost only with the
+   * run's log. Shared by every block's context copy and by child workflows billed on this log.
+   */
+  unrecordedModelCost?: UnrecordedModelCost
+}
+
+/** A run's running model spend, and when its usage limit may next be read after a read failed. */
+export interface UnrecordedModelCost {
+  total: number
+  readDeferredUntil?: number
+  /** Agent turns whose journaled spend `total` already holds. */
+  countedTurnIds?: Set<string>
 }
 
 /**

apps/sim/lib/billing/core/mid-run-usage.ts

@@ -89,8 +89,30 @@ function periodHasEnded(attribution: BillingAttributionSnapshot): boolean {
   return Date.now() >= new Date(attribution.billingPeriod.end).getTime()
 }
 
+/**
+ * The limit a run's not-yet-recorded spend takes it to, if any: the payer's pool, then the acting
+ * member's cap.
+ */
+function limitReachedWith(
+  usage: AttributedUsageLimitsResult,
+  unrecordedCost: number
+): 'payer' | 'member' | undefined {
+  // Without unrecorded spend the gate's own verdict stands, so callers that pass none are unchanged.
+  if (unrecordedCost <= 0) return undefined
+  const { payerUsage, memberUsage } = usage
+  if (payerUsage && payerUsage.currentUsage + unrecordedCost >= payerUsage.limit) return 'payer'
+  if (
+    memberUsage?.limit != null &&
+    memberUsage.currentUsage + unrecordedCost >= memberUsage.limit
+  ) {
+    return 'member'
+  }
+  return undefined
+}
+
 async function readGateVerdict(
-  attribution: BillingAttributionSnapshot
+  attribution: BillingAttributionSnapshot,
+  unrecordedCost: number
 ): Promise<MidRunUsageVerdict> {
   let usage: AttributedUsageLimitsResult
   try {
@@ -101,7 +123,10 @@ async function readGateVerdict(
     })
     return { status: 'unknown' }
   }
-  if (!usage.isExceeded) return { status: 'within' }
+  if (!usage.isExceeded) {
+    const scope = limitReachedWith(usage, unrecordedCost)
+    return scope ? { status: 'exceeded', scope } : { status: 'within' }
+  }
   if (usage.reason === 'billing_blocked') {
     return { status: 'blocked', ...(usage.message ? { message: usage.message } : {}) }
   }
@@ -120,12 +145,17 @@ async function readGateVerdict(
  * 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.
+ *
+ * @param unrecordedCost - Spend the ledger does not hold yet, counted as recorded usage.
  */
 export async function readMidRunUsageVerdict(
-  attribution: BillingAttributionSnapshot
+  attribution: BillingAttributionSnapshot,
+  unrecordedCost = 0
 ): Promise<MidRunUsageVerdict> {
   if (!isHosted || !isBillingEnabled) return { status: 'within' }
-  if (!rollsIntoCurrentPeriod(attribution.billingPeriod)) return readGateVerdict(attribution)
+  if (!rollsIntoCurrentPeriod(attribution.billingPeriod)) {
+    return readGateVerdict(attribution, unrecordedCost)
+  }
   let judged: BillingAttributionSnapshot
   try {
     judged = await currentAttribution(attribution)
@@ -136,7 +166,7 @@ export async function readMidRunUsageVerdict(
     return { status: 'unknown' }
   }
   if (periodHasEnded(judged)) return { status: 'unknown' }
-  const verdict = await readGateVerdict(judged)
+  const verdict = await readGateVerdict(judged, unrecordedCost)
   // A period that ended during the read is judged at the next callback, which reloads it.
   return periodHasEnded(judged) ? { status: 'unknown' } : verdict
 }

apps/sim/lib/memory/conversation-types.ts

@@ -91,6 +91,7 @@ export interface AgentConversationSession {
   getFinalResponse(): AgentTurnState['final']
   getFinalAssistantContent(): string | undefined
   readonly memoryId?: string
+  readonly turnId?: string
   getUsage(): ConversationUsageTotal
   recordContextUsage?(usage: ConversationUsageTotal): Promise<void>
   captureStep(step: CapturedConversationStep): Promise<void>

apps/sim/providers/anthropic/streaming-tool-loop.ts

@@ -585,8 +585,8 @@ export function createAnthropicStreamingToolLoopStream(
           aborted: loopAbortController.signal.aborted,
           consumerCancelled,
           error,
-          onUnexpectedError: (cause) =>
-            logger.error('Anthropic streaming tool loop failed', {
+          onUnexpectedError: (cause, level) =>
+            logger[level]('Anthropic streaming tool loop failed', {
               error: toError(cause).message,
             }),
         })

apps/sim/providers/bedrock/streaming-tool-loop.ts

@@ -698,8 +698,8 @@ export function createBedrockStreamingToolLoopStream(
           aborted: loopAbortController.signal.aborted,
           consumerCancelled,
           error,
-          onUnexpectedError: (cause) =>
-            logger.error('Bedrock streaming tool loop failed', {
+          onUnexpectedError: (cause, level) =>
+            logger[level]('Bedrock streaming tool loop failed', {
               error: toError(cause).message,
             }),
         })

apps/sim/providers/conversation-generation.test.ts

@@ -1,3 +1,8 @@
+import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing'
+import {
+  billingUsageGateCacheMock,
+  billingUsageGateCacheMockFns,
+} from '@sim/testing/mocks/billing-usage-gate-cache.mock'
 import {
   providersConversationHistoryMock,
   providersConversationHistoryMockFns,
@@ -29,6 +34,7 @@ const state = vi.hoisted(() => ({ enabled: true, historyTokens: 0 }))
 vi.mock('@/providers/conversation-history', () => providersConversationHistoryMock)
 vi.mock('@/providers/models', () => providersModelsMock)
 vi.mock('@/lib/tokenization/accurate', () => tokenizationAccurateMock)
+vi.mock('@/lib/billing/core/usage-gate-cache', () => billingUsageGateCacheMock)
 
 providersModelsMockFns.mockGetMaxOutputTokensForModel.mockReturnValue(100)
 Object.assign(providersModelsMock.PROVIDER_DEFINITIONS, {
@@ -611,3 +617,53 @@ describe('provider generation context boundary', () => {
     expect(payload.messages).toHaveLength(1)
   })
 })
+
+describe('usage limit across compaction', () => {
+  it('stops the generation when its compaction spends the run past the usage limit', async () => {
+    setEnvFlags({ isHosted: true, isBillingEnabled: true })
+    billingUsageGateCacheMockFns.mockCheckExecutionUsageLimits.mockResolvedValue({
+      isExceeded: false,
+      payerUsage: { currentUsage: 3, limit: 5 },
+    })
+    const unrecordedModelCost = { total: 1 }
+    providersConversationHistoryMockFns.mockGetConversationRequestContext.mockReturnValueOnce({
+      agentConversation: {},
+      agentMemoryContext: { historyTokens: 200 },
+      executionContext: {
+        metadata: {
+          duration: 0,
+          billingAttribution: {
+            actorUserId: 'actor',
+            workspaceId: 'workspace',
+            organizationId: null,
+            billedAccountUserId: 'actor',
+            billingEntity: { type: 'user', id: 'actor' },
+            billingPeriod: { start: '2026-10-01T00:00:00.000Z', end: '2099-01-01T00:00:00.000Z' },
+            payerSubscription: null,
+          },
+        },
+        unrecordedModelCost,
+      },
+    })
+    const fixture = fixtures[0]
+    const input = request()
+    bindConversationGenerationCompactor(input, async () => {
+      unrecordedModelCost.total += 1.5
+      return { role: 'user' as const, content: 'Derived prior context' }
+    })
+    const messages = [
+      fixture.prompt,
+      ...fixture.batch,
+      nativeText(fixture, 'old'.repeat(250)),
+      nativeText(fixture, 'latest receipt'),
+    ]
+
+    try {
+      await expect(
+        prepareConversationGeneration(input, fixture.protocol, { [fixture.key]: messages })
+      ).rejects.toThrow('usage limit')
+    } finally {
+      resetEnvFlagsMock()
+    }
+  })
+})