@@ -682,36 +682,79 @@ describe('async preprocessing correlation threading', () => {
     }
 
     it.each([
-      { statusCode: 404, message: 'Workflow not found' },
-      { statusCode: 403, message: 'Workflow is not deployed' },
-      { statusCode: 402, message: 'Usage limit exceeded. Please upgrade your plan to continue.' },
-      { statusCode: 429, message: 'Rate limit exceeded. Please try again later.' },
-    ])('completes the job for a non-retryable $statusCode', async ({ statusCode, message }) => {
+      { statusCode: 404, message: 'Workflow not found', workflowState: 'not_found' },
+      {
+        statusCode: 403,
+        message: 'Workflow is not deployed',
+        code: 'WORKFLOW_NOT_DEPLOYED',
+        workflowState: 'not_deployed',
+      },
+      {
+        statusCode: 403,
+        message:
+          'This workflow is not attached to a workspace. Personal workflows are deprecated and cannot execute.',
+        workflowState: 'not_attached',
+      },
+    ])('completes the job for a $statusCode $message that never started a log', async (error) => {
+      loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
+
+      await expect(runRejected(error)).resolves.toMatchObject({
+        success: false,
+        workflowId: 'workflow-1',
+        executionId: 'execution-rejected',
+        error: error.message,
+      })
+    })
+
+    const loggedRejections = [
+      {
+        statusCode: 402,
+        message: 'Usage limit exceeded. Please upgrade your plan to continue.',
+        code: 'USAGE_LIMIT_EXCEEDED',
+      },
+      {
+        statusCode: 402,
+        message: 'Usage headroom exhausted',
+        code: 'EXECUTION_PAYER_HEADROOM_EXHAUSTED',
+        retryable: false,
+      },
+      {
+        statusCode: 429,
+        message: 'Rate limit exceeded. Please try again later.',
+        code: 'RATE_LIMIT_EXCEEDED',
+      },
+      { statusCode: 403, message: 'Account suspended', code: 'ACCOUNT_SUSPENDED' },
+    ]
+
+    it.each(loggedRejections)('completes the job for a logged $code rejection', async (error) => {
       loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(true)
-      await expect(runRejected({ statusCode, message })).resolves.toMatchObject({
+
+      await expect(runRejected(error)).resolves.toMatchObject({
         success: false,
         workflowId: 'workflow-1',
         executionId: 'execution-rejected',
-        error: message,
+        error: error.message,
       })
       expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledWith(
-        expect.objectContaining({ error: expect.objectContaining({ message }) })
+        expect.objectContaining({ error: expect.objectContaining({ message: error.message }) })
       )
     })
 
-    it('fails the job when the rejection could not be logged', async () => {
-      loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
+    it.each(loggedRejections)(
+      'fails the job when a $code rejection could not be logged',
+      async (error) => {
+        loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
 
-      await expect(
-        runRejected({ statusCode: 403, message: 'Workflow is not deployed' })
-      ).rejects.toThrow('Workflow is not deployed')
-    })
+        await expect(runRejected(error)).rejects.toThrow(error.message)
+      }
+    )
 
     it.each([
       { statusCode: 429, message: 'Too many concurrent executions', retryable: true },
       { statusCode: 402, message: 'Usage is temporarily unavailable', retryable: true },
       { statusCode: 503, message: 'Usage admission is temporarily unavailable', retryable: true },
       { statusCode: 500, message: 'Internal error while fetching workflow' },
+      { statusCode: 404, message: 'Workflow not found' },
     ])('fails the job for $statusCode $message', async (error) => {
       await expect(runRejected(error)).rejects.toThrow(error.message)
     })
 
@@ -16,6 +16,8 @@ import {
   assertBillingAttributionSnapshot,
   type BillingAttributionSnapshot,
 } from '@/lib/billing/core/attribution'
+import { getDeterministicAdmissionRejectionCode } from '@/lib/core/admission/rejection'
+import { isReservationHeadroomDenial } from '@/lib/core/admission/transient-failure'
 import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import type { ExecutionScope } from '@/lib/core/errors/execution-scope-deleted'
@@ -30,7 +32,12 @@ import {
   RESERVATION_TTL_BUFFER_MS,
 } from '@/lib/core/execution-limits'
 import type { RequestAttribution } from '@/lib/core/utils/request-attribution'
-import { type PreprocessExecutionError, preprocessExecution } from '@/lib/execution/preprocessing'
+import {
+  isWorkflowStateRejection,
+  type PreprocessExecutionError,
+  preprocessExecution,
+  RATE_LIMIT_EXCEEDED_CODE,
+} from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
@@ -109,18 +116,23 @@ type LegacyWorkflowExecutionPayload = Omit<WorkflowExecutionPayload, 'principal'
 type WorkflowExecutionJobPayload = WorkflowExecutionPayload | LegacyWorkflowExecutionPayload
 
 /**
- * Preprocessing rejections that are an outcome of the workflow's or account's
- * state — deleted or undeployed workflow, suspended account, exhausted usage
- * or rate quota — rather than a fault. A retryable rejection (concurrency
- * capacity, an unreadable usage ledger, reservation infrastructure) and every
- * 5xx still fail the job.
+ * Whether a preprocessing rejection is an outcome of the workflow's or account's state rather
+ * than a fault. A workflow-state rejection writes no execution log, so it completes at once. A
+ * quota or account rejection (exhausted usage, spent headroom, rate limit, suspended account)
+ * writes one, so it completes only once that log is saved; an unsaved one (an outage) still fails
+ * the job and alerts. A retryable rejection and every other failure fail the job.
  */
-const NON_FAULT_PREPROCESSING_STATUS_CODES: ReadonlySet<number> = new Set([402, 403, 404, 429])
-
-function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError): boolean {
-  return (
-    NON_FAULT_PREPROCESSING_STATUS_CODES.has(rejection.statusCode) && rejection.retryable !== true
-  )
+function isNonFaultPreprocessingRejection(
+  rejection: PreprocessExecutionError,
+  rejectionLogged: boolean
+): boolean {
+  if (rejection.retryable === true) return false
+  if (isWorkflowStateRejection(rejection)) return true
+  const quotaRejection =
+    getDeterministicAdmissionRejectionCode(rejection) !== undefined ||
+    isReservationHeadroomDenial(rejection) ||
+    rejection.code === RATE_LIMIT_EXCEEDED_CODE
+  return quotaRejection && rejectionLogged
 }
 
 /** Completes the job with a failure already in the execution log, keeping it on the run's span. */
@@ -378,11 +390,9 @@ export async function executeWorkflowJob(
         const scopeDeletedOnCompletion = loggingSession.getDeletedScope()
         if (scopeDeletedOnCompletion) return completeWithDeletedScope(scopeDeletedOnCompletion)
 
-        // Complete only when the rejection is durably logged; an unsaved one must still alert.
         if (
           preprocessingRejection &&
-          isNonFaultPreprocessingRejection(preprocessingRejection) &&
-          loggingSession.hasCompleted()
+          isNonFaultPreprocessingRejection(preprocessingRejection, loggingSession.hasCompleted())
         ) {
           return completeWithRecordedFailure(workflowId, executionId, error)
         }
 
@@ -262,7 +262,7 @@ export async function handleNewSignInAccount(
 ): Promise<void> {
   if (!isUserCreatedInAuthRequest(userId, authRequest)) return
   if (!(await isDeletedAccountIdentity({ oauthSubjects: [{ providerId, accountId }] }))) return
-  await forfeitFreeTierAllowance(userId)
+  await forfeitFreeTierAllowance(db, userId)
 }
 
 /**
@@ -283,29 +283,45 @@ export async function ensureUserStatsExists(
   userId: string,
   authRequest?: object | null
 ): Promise<void> {
-  const created = await db
+  await createUserStatsIfMissing(db, userId, {
+    includeSignIns: isUserCreatedInAuthRequest(userId, authRequest),
+  })
+}
+
+/**
+ * Creates a user's `user_stats` row with the free allowance when none exists, on `executor`
+ * (the caller's transaction, or `db`). Only a row this call creates gets the deleted-identity
+ * check signup makes: the email always, and the linked sign-ins only when `includeSignIns` is
+ * set. An existing row costs only the conflicting insert. A failed check leaves the allowance in
+ * place and never fails the caller or aborts its transaction.
+ *
+ * @returns whether this call created the row
+ */
+export async function createUserStatsIfMissing(
+  executor: DbOrTx,
+  userId: string,
+  { includeSignIns = false }: { includeSignIns?: boolean } = {}
+): Promise<boolean> {
+  const created = await executor
     .insert(userStats)
     .values({
       id: generateId(),
-      userId: userId,
+      userId,
       currentUsageLimit: getFreeTierLimit().toString(),
       usageLimitUpdatedAt: new Date(),
     })
     .onConflictDoNothing({ target: userStats.userId })
     .returning({ id: userStats.id })
-  if (created.length === 0) return
+  if (created.length === 0) return false
 
   try {
-    const includeSignIns = isUserCreatedInAuthRequest(userId, authRequest)
-    if (await userHasDeletedAccountIdentity(userId, { includeSignIns })) {
-      await forfeitFreeTierAllowance(userId)
+    if (await userHasDeletedAccountIdentity(userId, { includeSignIns }, executor)) {
+      await forfeitFreeTierAllowance(executor, userId)
     }
   } catch (error) {
-    logger.error('Failed to withhold the free allowance from a fallback stats row', {
-      userId,
-      error,
-    })
+    logger.error('Failed to withhold the free allowance from a new stats row', { userId, error })
   }
+  return true
 }
 
 interface ResolvedUserUsageData {
@@ -690,12 +706,12 @@ export async function checkUsageStatus(userId: string): Promise<{
 /**
  * Withholds the free allowance from a new account found to reuse the sign-in
  * identity of a recently deleted one after its stats row was created. The flag
- * and the limit it implies commit together, so a failed re-derivation can never
- * leave the account flagged but still holding the free limit; repeating the
- * call is harmless.
+ * and the limit it implies commit together (a transaction on `db`, a savepoint
+ * inside a caller's transaction), so a failed re-derivation can never leave the
+ * account flagged but still holding the free limit; repeating the call is harmless.
  */
-async function forfeitFreeTierAllowance(userId: string): Promise<void> {
-  await db.transaction(async (tx) => {
+async function forfeitFreeTierAllowance(executor: DbOrTx, userId: string): Promise<void> {
+  await executor.transaction(async (tx) => {
     await tx.update(userStats).set({ freeTierEligible: false }).where(eq(userStats.userId, userId))
     await syncUsageLimitsFromSubscription(userId, tx)
   })
 
@@ -29,7 +29,7 @@ import {
   invalidateSecurityPolicyVersionCache,
 } from '@/lib/auth/security-policy'
 import { applySessionPolicyToNewMember } from '@/lib/auth/session-policy'
-import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
+import { createUserStatsIfMissing, syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
 import {
   assertNoUnresolvedEnterpriseIssuance,
   resolveEnterpriseMetadataIntent,
@@ -1935,14 +1935,7 @@ export async function transferOrganizationOwnership(
         .limit(1)
 
       if (oldStats) {
-        await tx
-          .insert(userStats)
-          .values({
-            id: generateId(),
-            userId: newOwnerUserId,
-            usageLimitUpdatedAt: new Date(),
-          })
-          .onConflictDoNothing({ target: userStats.userId })
+        await createUserStatsIfMissing(tx, newOwnerUserId)
 
         const overage = oldStats.billedOverageThisPeriod || '0'
         const overageNum = toNumber(toDecimal(overage))
 
@@ -19,11 +19,11 @@
 
 import { organization, project, userStats, workspace } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
 import { and, asc, eq, inArray, sql } from 'drizzle-orm'
 import { maybeNotifyLimit } from '@/lib/billing/core/limit-notifications'
 import type { HighestPrioritySubscription } from '@/lib/billing/core/plan'
+import { createUserStatsIfMissing } from '@/lib/billing/core/usage'
 import type { BillingEntity } from '@/lib/billing/core/usage-log'
 import type {
   ProjectStorageBillingContext,
@@ -39,7 +39,7 @@ import {
   isStorageEnforcementEnabled,
   StorageLimitExceededError,
 } from '@/lib/billing/storage/limits'
-import { getFreeTierLimit, isOrgScopedSubscription } from '@/lib/billing/subscriptions/utils'
+import { isOrgScopedSubscription } from '@/lib/billing/subscriptions/utils'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { lockProject } from '@/lib/projects/membership'
 
@@ -764,14 +764,12 @@ export async function incrementAdmittedStorageUsageForBillingContextInTx(
  * window between the read and the increment. This helper does not send
  * notifications because it runs mid-transaction.
  *
- * For a personal (non-org-scoped) `userId`, this first upserts the
- * `userStats` row on `tx` — a documented possibility for OAuth account
- * linking (see `ensureUserStatsExists` in `lib/billing/core/usage.ts`, whose
- * insert values this mirrors) — because the conditional `UPDATE` below
- * matches 0 rows, and therefore reads as "quota exceeded", if that row
- * doesn't exist yet. `ensureUserStatsExists` itself isn't reused here since
- * it writes through the standalone `db` client, which would open a second
- * pooled connection while this transaction's is held.
+ * For a personal (non-org-scoped) `userId`, this first creates a missing
+ * `userStats` row on `tx` with `createUserStatsIfMissing` — a documented
+ * possibility for OAuth account linking — because the conditional `UPDATE`
+ * below matches 0 rows, and therefore reads as "quota exceeded", if that row
+ * doesn't exist yet. It runs on `tx` rather than the standalone `db` client,
+ * which would open a second pooled connection while this transaction's is held.
  */
 export async function checkAndIncrementStorageUsageInTx(
   tx: DbOrTx,
@@ -791,17 +789,7 @@ export async function checkAndIncrementStorageUsageInTx(
 
   const orgScoped = isOrgScopedSubscription(sub, userId) && sub
 
-  if (!orgScoped) {
-    await tx
-      .insert(userStats)
-      .values({
-        id: generateId(),
-        userId,
-        currentUsageLimit: getFreeTierLimit().toString(),
-        usageLimitUpdatedAt: new Date(),
-      })
-      .onConflictDoNothing({ target: userStats.userId })
-  }
+  if (!orgScoped) await createUserStatsIfMissing(tx, userId)
 
   const [updated] = orgScoped
     ? await tx
 
@@ -83,6 +83,23 @@ function hasAdmissionCode(failure: AdmissionFailureLike, code: string): boolean
   return failure.code === code || failure.cause?.code === code
 }
 
+const RESERVATION_HEADROOM_DESCRIPTORS = [
+  ADMISSION_ERROR_DESCRIPTOR.RESERVATION_PAYER_HEADROOM,
+  ADMISSION_ERROR_DESCRIPTOR.RESERVATION_MEMBER_HEADROOM,
+] as const
+
+/**
+ * Whether a reservation was denied because the payer's or member's usage headroom is spent. The
+ * run is refused for good, though a later one can be admitted as in-flight runs settle, which is
+ * why deterministic admission rejections leave these out.
+ */
+export function isReservationHeadroomDenial(failure: AdmissionFailureLike): boolean {
+  return RESERVATION_HEADROOM_DESCRIPTORS.some(
+    (descriptor) =>
+      failure.statusCode === descriptor.statusCode && hasAdmissionCode(failure, descriptor.code)
+  )
+}
+
 /**
  * Identifies only admission failures that unattended callers can safely retry.
  */
 
@@ -338,9 +338,12 @@ export interface JobQueueBackend {
   cancelByKey(cancelKey: string): boolean
 
   /**
-   * Cancel every queued, delayed or running workflow job (`WORKSPACE_EXECUTION_JOB_TYPES`)
-   * enqueued for a workspace, for when the workspace itself is being removed. Throws when any
-   * listing or cancellation failed, after attempting the rest, so the caller can retry.
+   * Cancel every queued, delayed or running persisted workflow job
+   * (`WORKSPACE_EXECUTION_JOB_TYPES`) enqueued for a workspace, for when the workspace itself is
+   * being removed. Runs the database backend executes through `batchEnqueueAndWait` keep no job
+   * row, so this never reaches them; they stop through their execution's cancellation signal.
+   * Throws when any listing or cancellation failed, after attempting the rest, so the caller can
+   * retry.
    */
   cancelByWorkspace(
     workspaceId: string,