@@ -3,6 +3,7 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { purgeExpiredDeletedAccountIdentities } from '@/lib/users/deleted-identity'
 import { runCleanupOAuthTokens } from '@/background/cleanup-oauth-tokens'
 
 export const dynamic = 'force-dynamic'
@@ -14,14 +15,22 @@ const logger = createLogger('CleanupOAuthTokensAPI')
  * Retention sweep for lapsed OAuth token rows. Issuance state does not gate
  * retention: a deployment that disables the provider must still drain rows
  * created while it was enabled.
+ *
+ * The same hourly run drains expired deleted-account identity digests first,
+ * within its own short budget, so they never outlive their retention window. A
+ * failed purge is logged and never holds up the token sweep.
  */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   const authError = verifyCronAuth(request, 'OAuth token cleanup')
   if (authError) return authError
 
   try {
+    const deletedAccountIdentities = await purgeExpiredDeletedAccountIdentities().catch((error) => {
+      logger.error('Failed to purge expired deleted-account identities', { error })
+      return null
+    })
     const result = await runCleanupOAuthTokens()
-    return NextResponse.json({ success: true, ...result })
+    return NextResponse.json({ success: true, ...result, deletedAccountIdentities })
   } catch (error) {
     logger.error('Failed to sweep expired OAuth tokens', { error })
     return NextResponse.json(
 
@@ -21,6 +21,7 @@ const hoistedMocks = vi.hoisted(() => ({
   billingSubscriptions: [] as unknown[],
   idempotencyCalls: [] as { namespace: string; requestFingerprint: string }[],
   moveWorkspace: vi.fn(),
+  userHasDeletedAccountIdentity: vi.fn(async () => false),
 }))
 
 vi.mock('@sim/audit', () => auditMock)
@@ -53,6 +54,9 @@ vi.mock('@/lib/billing/enterprise-outbox', () => ({
   resolveEnterpriseMetadataIntent: vi.fn(),
 }))
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
+vi.mock('@/lib/users/deleted-identity', () => ({
+  userHasDeletedAccountIdentity: hoistedMocks.userHasDeletedAccountIdentity,
+}))
 
 import { grantDashboardOrganizationBalance, grantDashboardUserBalance } from '@/lib/admin/dashboard'
 
@@ -150,6 +154,30 @@ describe('grantDashboardUserBalance', () => {
     )
   })
 
+  it('creates a missing stats row without the free allowance for a returning deleted identity', async () => {
+    mocks.userHasDeletedAccountIdentity.mockResolvedValueOnce(true)
+    queueTableRows(user, [{ id: 'user-1' }])
+    queueTableRows(userStats, [
+      { creditBalance: '0', currentUsageLimit: '0', freeTierEligible: false },
+    ])
+    mocks.billingSubscriptions = [null, null]
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      { creditBalance: '0.005', currentUsageLimit: '0.005' },
+    ])
+
+    await grantDashboardUserBalance(
+      'user-1',
+      0.005,
+      undefined,
+      '4f0f5c4e-3a43-4f39-9d55-6b1b2f6f2a10',
+      { id: 'admin-1', name: 'Admin', email: 'admin@sim.ai' }
+    )
+
+    expect(dbChainMockFns.values).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'user-1', freeTierEligible: false, currentUsageLimit: '0' })
+    )
+  })
+
   it('adds above a higher custom limit for an entitled personal subscription', async () => {
     const personalSubscription = {
       referenceId: 'user-1',
 
@@ -78,6 +78,7 @@ import { executeTransactionallyIdempotent } from '@/lib/core/idempotency/transac
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 import type { DbOrTx } from '@/lib/db/types'
 import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
+import { userHasDeletedAccountIdentity } from '@/lib/users/deleted-identity'
 import { ownedAttachableWorkspacesWhere } from '@/lib/workspaces/organization-workspaces'
 
 interface PaginationInput {
@@ -1497,6 +1498,14 @@ export async function grantDashboardUserBalance(
   actor: AdminMutationActor
 ) {
   const normalizedReason = reason?.trim() || null
+  /**
+   * Decides the eligibility of a stats row this grant may have to create, with
+   * the same email check as every other row creation. Looked up before the
+   * transaction so it never holds a second pooled connection inside it.
+   */
+  const freeTierEligibleIfCreated = !(await userHasDeletedAccountIdentity(userId, {
+    includeSignIns: false,
+  }))
   const outcome = await db.transaction(async (tx) => {
     await acquireUserBillingIdentityLock(tx, userId)
     const [account] = await tx
@@ -1515,17 +1524,17 @@ export async function grantDashboardUserBalance(
       .from(member)
       .where(eq(member.userId, userId))
       .limit(1)
-    // Only inserts when signup never created the row; that row keeps the column's default eligibility.
     const initialUsageLimit =
       initialMembership || isOrgScopedSubscription(initialSubscription, userId)
         ? null
-        : getPerUserMinimumLimit(initialSubscription, true).toString()
+        : getPerUserMinimumLimit(initialSubscription, freeTierEligibleIfCreated).toString()
     await tx
       .insert(userStats)
       .values({
         id: generateId(),
         userId,
         currentUsageLimit: initialUsageLimit,
+        freeTierEligible: freeTierEligibleIfCreated,
         usageLimitUpdatedAt: new Date(),
       })
       .onConflictDoNothing({ target: userStats.userId })
 
@@ -8,6 +8,7 @@ import { account, deletedAccountIdentity, user, userStats } from '@sim/db/schema
 import { authMock } from '@sim/testing/mocks/auth.mock'
 import { envFlagsMock } from '@sim/testing/mocks/env-flags.mock'
 import { generateId } from '@sim/utils/id'
+import { generateRandomHex } from '@sim/utils/random'
 import { eq, inArray } from 'drizzle-orm'
 import { afterAll, describe, expect, it, vi } from 'vitest'
 import {
@@ -18,6 +19,7 @@ import {
 } from '@/lib/billing/core/usage'
 import { getFreeTierLimit } from '@/lib/billing/subscriptions/utils'
 import { deleteUserAccount } from '@/lib/users/account-deletion'
+import { purgeExpiredDeletedAccountIdentities } from '@/lib/users/deleted-identity'
 
 vi.mock('@/lib/core/config/env-flags', () => ({
   ...envFlagsMock,
@@ -34,15 +36,15 @@ interface SignIn {
   accountId: string
 }
 
-async function insertUser(email: string): Promise<string> {
+async function insertUser(email: string, { emailVerified = true } = {}): Promise<string> {
   const id = generateId()
   userIds.push(id)
   const now = new Date()
   await db.insert(user).values({
     id,
     name: 'Deleted identity fixture',
     email,
-    emailVerified: true,
+    emailVerified,
     createdAt: now,
     updatedAt: now,
   })
@@ -290,6 +292,52 @@ describe('deleted account identities and the free allowance', () => {
     expect(Math.abs(rows[0].expiresAt.getTime() - (deletedAt + thirtyDays))).toBeLessThan(60_000)
   })
 
+  it('never records an unverified email, so its real owner keeps the allowance', async () => {
+    const email = `unverified.${generateId()}@deleted-identity.test`
+    const github = { providerId: 'github', accountId: `gh-${generateId()}` }
+    const squatter = await insertUser(email, { emailVerified: false })
+    await linkAccount(squatter, github)
+    await handleNewUser(squatter, {})
+    const written = await deleteOwnAccount(squatter)
+
+    const owner = await signUp(email)
+    const sameGithub = await signUp(`unverified-gh.${generateId()}@deleted-identity.test`, github)
+
+    expect(written).toHaveLength(1)
+    expect(await statsOf(owner)).toEqual({ freeTierEligible: true, limit: getFreeTierLimit() })
+    expect(await statsOf(sameGithub)).toEqual({ freeTierEligible: false, limit: 0 })
+  })
+
+  it('drains every expired digest on the scheduled purge and keeps live ones', async () => {
+    const expired = Array.from({ length: 2_345 }, () => generateRandomHex(64))
+    const live = Array.from({ length: 3 }, () => generateRandomHex(64))
+    recordedHashes.push(...expired, ...live)
+    const past = new Date(Date.now() - 60_000)
+    const future = new Date(Date.now() + 60 * 60_000)
+    for (let start = 0; start < expired.length; start += 500) {
+      await db
+        .insert(deletedAccountIdentity)
+        .values(
+          expired
+            .slice(start, start + 500)
+            .map((identityHash) => ({ identityHash, expiresAt: past }))
+        )
+    }
+    await db
+      .insert(deletedAccountIdentity)
+      .values(live.map((identityHash) => ({ identityHash, expiresAt: future })))
+
+    const result = await purgeExpiredDeletedAccountIdentities()
+
+    const remaining = await db
+      .select({ hash: deletedAccountIdentity.identityHash })
+      .from(deletedAccountIdentity)
+      .where(inArray(deletedAccountIdentity.identityHash, [...expired, ...live]))
+    expect(result.drained).toBe(true)
+    expect(result.deleted).toBeGreaterThanOrEqual(expired.length)
+    expect(remaining.map((row) => row.hash).sort()).toEqual([...live].sort())
+  })
+
   it('records only sign-in identities, as keyed digests holding no address or subject', async () => {
     const email = `digest.${generateId()}@deleted-identity.test`
     const githubSubject = `gh-${generateId()}`
 
@@ -13,7 +13,7 @@ describe('recordDeletedAccountIdentities', () => {
   beforeEach(() => {
     resetDbChainMock()
     setEnvFlags({ isBillingEnabled: true })
-    queueTableRows(schemaMock.user, [{ email: 'ada@example.com' }])
+    queueTableRows(schemaMock.user, [{ email: 'ada@example.com', emailVerified: true }])
   })
 
   afterEach(() => {
 
@@ -4,7 +4,7 @@ import { createLogger } from '@sim/logger'
 import { hmacSha256Hex } from '@sim/security/hmac'
 import { getErrorMessage } from '@sim/utils/errors'
 import { normalizeEmail } from '@sim/utils/string'
-import { and, eq, gt, inArray, lte, sql } from 'drizzle-orm'
+import { and, asc, eq, gt, inArray, lte, sql } from 'drizzle-orm'
 import { env } from '@/lib/core/config/env'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import type { DbOrTx } from '@/lib/db/types'
@@ -42,9 +42,18 @@ const logger = createLogger('DeletedAccountIdentity')
  */
 const DELETED_IDENTITY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000
 
-/** Expired rows removed per recorded deletion, which keeps the table bounded without a cron. */
+/** Expired rows a recorded deletion removes opportunistically on its way through. */
 const EXPIRED_PURGE_BATCH = 100
 
+/** Rows per batch of the scheduled purge, each its own short transaction. */
+const SCHEDULED_PURGE_BATCH = 1_000
+
+/** Batches one scheduled run may delete before leaving the rest to the next run. */
+const SCHEDULED_PURGE_MAX_BATCHES = 50
+
+/** Stops admitting batches well inside the cron request's 60-second timeout. */
+const SCHEDULED_PURGE_BUDGET_MS = 10_000
+
 const GMAIL_DOMAINS = new Set(['gmail.com', 'googlemail.com'])
 
 /**
@@ -129,10 +138,13 @@ function identityHashes(identity: SignInIdentity): string[] {
   return [...new Set(preimages)].map((preimage) => hmacSha256Hex(preimage, key))
 }
 
-/** A user's email and the social sign-in subjects currently linked to it. */
-async function loadSignInIdentity(executor: DbOrTx, userId: string): Promise<SignInIdentity> {
+/** A user's email (and whether it is verified) and the social sign-in subjects linked to it. */
+async function loadSignInIdentity(
+  executor: DbOrTx,
+  userId: string
+): Promise<SignInIdentity & { emailVerified: boolean }> {
   const [profile] = await executor
-    .select({ email: user.email })
+    .select({ email: user.email, emailVerified: user.emailVerified })
     .from(user)
     .where(eq(user.id, userId))
     .limit(1)
@@ -145,14 +157,20 @@ async function loadSignInIdentity(executor: DbOrTx, userId: string): Promise<Sig
         inArray(account.providerId, [...RECORDED_SIGN_IN_PROVIDER_IDS])
       )
     )
-  return { email: profile?.email, oauthSubjects }
+  return { email: profile?.email, emailVerified: profile?.emailVerified === true, oauthSubjects }
 }
 
 /**
  * Records the sign-in identities of an account being deleted at its holder's
  * request. Runs inside the deletion transaction, before the `user` row and its
  * `account` rows cascade away, so the record and the erasure commit together.
  *
+ * The email is recorded only when it was verified: an unverified address may
+ * belong to someone else, who must not lose the allowance for an account they
+ * never held. Its GitHub and Microsoft subjects are recorded either way. Where
+ * email verification is off, password accounts are therefore recorded by
+ * nothing (fails open).
+ *
  * The write runs in a savepoint and never fails the deletion: an anti-abuse
  * record must not become a way to block an erasure request. On failure the
  * savepoint rolls back, nothing is recorded, and the deletion proceeds.
@@ -162,7 +180,8 @@ export async function recordDeletedAccountIdentities(tx: DbOrTx, userId: string)
 
   try {
     await tx.transaction(async (savepoint) => {
-      const hashes = identityHashes(await loadSignInIdentity(savepoint, userId))
+      const { email, emailVerified, oauthSubjects } = await loadSignInIdentity(savepoint, userId)
+      const hashes = identityHashes({ email: emailVerified ? email : null, oauthSubjects })
       if (hashes.length === 0) return
 
       await savepoint.delete(deletedAccountIdentity).where(
@@ -242,3 +261,40 @@ export async function userHasDeletedAccountIdentity(
     return false
   }
 }
+
+/**
+ * Deletes every expired identity digest, so none outlives its retention window
+ * even when no later deletion comes along to purge it. Drains in short batches
+ * in `expires_at` order, skipping rows a concurrent deletion has locked, until
+ * none remain, the batch cap is hit, or the time budget runs out; whatever is
+ * left goes to the next run. Runs whether or not billing is enabled, since
+ * deleting expired rows is always safe.
+ */
+export async function purgeExpiredDeletedAccountIdentities(): Promise<{
+  deleted: number
+  drained: boolean
+}> {
+  const deadline = Date.now() + SCHEDULED_PURGE_BUDGET_MS
+  let deleted = 0
+  for (let batch = 0; batch < SCHEDULED_PURGE_MAX_BATCHES; batch += 1) {
+    if (Date.now() >= deadline) return { deleted, drained: false }
+    const removed = await db
+      .delete(deletedAccountIdentity)
+      .where(
+        inArray(
+          deletedAccountIdentity.identityHash,
+          db
+            .select({ identityHash: deletedAccountIdentity.identityHash })
+            .from(deletedAccountIdentity)
+            .where(lte(deletedAccountIdentity.expiresAt, sql`now()`))
+            .orderBy(asc(deletedAccountIdentity.expiresAt))
+            .limit(SCHEDULED_PURGE_BATCH)
+            .for('update', { skipLocked: true })
+        )
+      )
+      .returning({ identityHash: deletedAccountIdentity.identityHash })
+    deleted += removed.length
+    if (removed.length < SCHEDULED_PURGE_BATCH) return { deleted, drained: true }
+  }
+  return { deleted, drained: false }
+}