@@ -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(() => {})
       }
 
@@ -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
 
@@ -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 }>
 
@@ -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: [],
 
@@ -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({
 
@@ -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.
 
@@ -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'
 
@@ -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) {
 
@@ -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>
 }
 
 /**
 
@@ -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
 }