diff --git a/apps/sim/app/api/files/serve/[...path]/route.ts b/apps/sim/app/api/files/serve/[...path]/route.ts
index 478bab67a90..ec99bbfb84b 100644
--- a/apps/sim/app/api/files/serve/[...path]/route.ts
+++ b/apps/sim/app/api/files/serve/[...path]/route.ts
@@ -389,6 +389,7 @@ export const GET = withRouteHandler(
         request.method === 'HEAD'
       )
     } catch (error) {
+      if (request.signal.aborted) throw error
       if (error instanceof InternalUnauthenticatedError) {
         logger.warn('Unauthorized file access attempt', { error: error.message })
         return NextResponse.json(
@@ -599,6 +600,7 @@ async function handleLocalFile(
       options.ifNoneMatch
     )
   } catch (error) {
+    if (signal?.aborted) throw error
     logServeFailure('Error reading local file:', error)
     throw error
   }
@@ -689,6 +691,7 @@ async function handleCloudProxy(
       options.ifNoneMatch
     )
   } catch (error) {
+    if (signal?.aborted) throw error
     logServeFailure('Error downloading from cloud storage:', error)
     throw error
   }
diff --git a/apps/sim/lib/core/outbox/handlers.ts b/apps/sim/lib/core/outbox/handlers.ts
index 77c088f6b52..e4596596e33 100644
--- a/apps/sim/lib/core/outbox/handlers.ts
+++ b/apps/sim/lib/core/outbox/handlers.ts
@@ -44,6 +44,10 @@ import {
   WORKSPACE_FILE_LIVE_DOC_OUTBOX_EVENT,
   WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT,
 } from '@/lib/uploads/contexts/workspace/file-outbox-events'
+import {
+  ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT,
+  ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+} from '@/lib/users/storage-sweep-events'
 import { WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS } from '@/lib/workflows/deployment-outbox-events'
 import { MIGRATED_INVITATION_EMAIL_EVENT_TYPE } from '@/lib/workspaces/admin-move-event'
 import {
@@ -231,6 +235,13 @@ export const OUTBOX_HANDLER_GROUPS: readonly LazyOutboxHandlerGroup[] = [
       return projectFilePrefixCleanupOutboxHandlers
     },
   },
+  {
+    events: [ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT, ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT],
+    load: async () => {
+      const { accountStorageSweepOutboxHandlers } = await import('@/lib/users/storage-sweep')
+      return accountStorageSweepOutboxHandlers
+    },
+  },
   {
     events: [PROJECT_FILE_DOCUMENT_RETIRE_EVENT],
     load: async () => {
diff --git a/apps/sim/lib/execution/payloads/store.local.test.ts b/apps/sim/lib/execution/payloads/store.local.test.ts
new file mode 100644
index 00000000000..38f061364da
--- /dev/null
+++ b/apps/sim/lib/execution/payloads/store.local.test.ts
@@ -0,0 +1,80 @@
+import { mkdtemp, readdir, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import {
+  largeValueMetadataMock,
+  largeValueMetadataMockFns,
+} from '@sim/testing/mocks/large-value-metadata.mock'
+import { uploadsConfigMock } from '@sim/testing/mocks/uploads-config.mock'
+import {
+  uploadsMetadataMock,
+  uploadsMetadataMockFns,
+} from '@sim/testing/mocks/uploads-metadata.mock'
+import { setUploadDirServer, uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock'
+import { describeError } from '@sim/utils/errors'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock)
+vi.mock('@/lib/uploads/config', () => uploadsConfigMock)
+vi.mock('@/lib/uploads/server/metadata', () => uploadsMetadataMock)
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => largeValueMetadataMock)
+
+import { storeLargeValue } from '@/lib/execution/payloads/store'
+
+const { mockInsertFileMetadata, mockDeleteFileMetadata } = uploadsMetadataMockFns
+const scope = { workspaceId: 'workspace-1', workflowId: 'workflow-1', executionId: 'execution-1' }
+
+let directory: string
+
+async function storedObjects(): Promise<string[]> {
+  try {
+    return await readdir(join(directory, 'execution', ...Object.values(scope)))
+  } catch (error) {
+    if (describeError(error).code === 'ENOENT') return []
+    throw error
+  }
+}
+
+describe('large value persistence against real local storage', () => {
+  beforeEach(async () => {
+    directory = await mkdtemp(join(tmpdir(), 'sim-large-value-'))
+    setUploadDirServer(directory)
+    mockInsertFileMetadata.mockReset().mockResolvedValue({ id: 'file-1' })
+    mockDeleteFileMetadata.mockReset().mockResolvedValue(true)
+    largeValueMetadataMockFns.mockRegisterLargeValueOwner.mockResolvedValue(true)
+  })
+
+  afterEach(async () => {
+    await rm(directory, { recursive: true, force: true })
+  })
+
+  it('removes the uploaded object when its metadata row cannot be written', async () => {
+    mockInsertFileMetadata.mockRejectedValueOnce(new Error('workspace no longer exists'))
+    const json = JSON.stringify({ payload: 'x'.repeat(2048) })
+
+    await expect(
+      storeLargeValue(JSON.parse(json), json, Buffer.byteLength(json), {
+        ...scope,
+        userId: 'user-1',
+        requireDurable: true,
+      })
+    ).rejects.toThrow('Failed to persist large execution value')
+
+    expect(await storedObjects()).toEqual([])
+    expect(largeValueMetadataMockFns.mockRegisterLargeValueOwner).not.toHaveBeenCalled()
+  })
+
+  it('keeps the object once its metadata row is written', async () => {
+    const json = JSON.stringify({ payload: 'y'.repeat(2048) })
+
+    const ref = await storeLargeValue(JSON.parse(json), json, Buffer.byteLength(json), {
+      ...scope,
+      userId: 'user-1',
+      requireDurable: true,
+    })
+
+    expect((await storedObjects()).filter((name) => name.endsWith('.json'))).toEqual([
+      ref.key?.split('/').pop(),
+    ])
+  })
+})
diff --git a/apps/sim/lib/execution/payloads/store.ts b/apps/sim/lib/execution/payloads/store.ts
index 0aedf8bc19e..b59416371cf 100644
--- a/apps/sim/lib/execution/payloads/store.ts
+++ b/apps/sim/lib/execution/payloads/store.ts
@@ -90,6 +90,7 @@ async function persistValue(
       context: 'execution',
       preserveKey: true,
       customKey: key,
+      cleanupOnMetadataFailure: true,
       metadata: {
         originalName: `large-value-${id}.json`,
         uploadedAt: new Date().toISOString(),
diff --git a/apps/sim/lib/projects/files/application/uploads.ts b/apps/sim/lib/projects/files/application/uploads.ts
index b3d41dab88a..188904907a1 100644
--- a/apps/sim/lib/projects/files/application/uploads.ts
+++ b/apps/sim/lib/projects/files/application/uploads.ts
@@ -29,10 +29,7 @@ import {
   prepareProjectFileAccounting,
   recordProjectFileWriteEffects,
 } from '@/lib/projects/files/application/write-effects'
-import {
-  projectUploadCleanupAvailableAt,
-  queueRetiredProjectUploadCleanup,
-} from '@/lib/projects/files/prefix-cleanup'
+import { queueRetiredProjectUploadCleanup } from '@/lib/projects/files/prefix-cleanup'
 import { resolveFileFolderTarget } from '@/lib/uploads/contexts/workspace/file-folder-manager'
 import {
   adoptVerifiedUploadSession,
@@ -43,6 +40,7 @@ import {
 } from '@/lib/uploads/contexts/workspace/file-manager'
 import type { WorkspaceFileSecretProvenance } from '@/lib/uploads/contexts/workspace/file-secret-provenance'
 import { enqueueWorkspaceFileStorageCleanups } from '@/lib/uploads/contexts/workspace/file-storage-cleanup-outbox'
+import { storageCleanupAvailableAt } from '@/lib/uploads/core/prefix-sweep'
 import { downloadFile } from '@/lib/uploads/core/storage-service'
 import { requestOrigin } from '@/lib/uploads/upload-session/application'
 import { assertProjectFileUploadBinding } from '@/lib/uploads/upload-session/project-file-binding'
@@ -321,7 +319,7 @@ const registerProjectUpload = defineAuthorizedProjectFileUseCase({
     await registration.registerFile(file.id)
     const cleanupIds = prepared.restored
       ? await enqueueWorkspaceFileStorageCleanups(tx, [input.session.finalKey], 'project', {
-          availableAt: projectUploadCleanupAvailableAt(),
+          availableAt: storageCleanupAvailableAt(),
         })
       : []
     const result = await mapProjectFileResult(tx, context, file)
diff --git a/apps/sim/lib/projects/files/prefix-cleanup.test.ts b/apps/sim/lib/projects/files/prefix-cleanup.test.ts
new file mode 100644
index 00000000000..0305976c60a
--- /dev/null
+++ b/apps/sim/lib/projects/files/prefix-cleanup.test.ts
@@ -0,0 +1,79 @@
+import { mkdir, mkdtemp, rm, stat, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { dirname, join } from 'node:path'
+import { resetDbChainMock } from '@sim/testing'
+import { uploadsConfigMock } from '@sim/testing/mocks/uploads-config.mock'
+import { uploadsMetadataMock } from '@sim/testing/mocks/uploads-metadata.mock'
+import { setUploadDirServer, uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock)
+vi.mock('@/lib/uploads/config', () => uploadsConfigMock)
+vi.mock('@/lib/uploads/server/metadata', () => uploadsMetadataMock)
+
+import type { OutboxEventContext } from '@/lib/core/outbox/service'
+import { PROJECT_FILE_PREFIX_CLEANUP_EVENT } from '@/lib/projects/files/outbox-events'
+import { projectFilePrefixCleanupOutboxHandlers } from '@/lib/projects/files/prefix-cleanup'
+
+const cleanup = projectFilePrefixCleanupOutboxHandlers[PROJECT_FILE_PREFIX_CLEANUP_EVENT]
+
+let root: string
+
+function outboxContext(deadlineAt?: number): OutboxEventContext {
+  return {
+    eventId: 'event-1',
+    eventType: PROJECT_FILE_PREFIX_CLEANUP_EVENT,
+    attempts: 0,
+    maxAttempts: 10,
+    signal: new AbortController().signal,
+    ...(deadlineAt ? { deadlineAt } : {}),
+    checkpointPayload: vi.fn(async () => {}),
+  }
+}
+
+async function put(key: string): Promise<void> {
+  await mkdir(dirname(join(root, key)), { recursive: true })
+  await writeFile(join(root, key), 'x')
+}
+
+async function exists(key: string): Promise<boolean> {
+  try {
+    await stat(join(root, key))
+    return true
+  } catch {
+    return false
+  }
+}
+
+describe('Project prefix cleanup on local storage', () => {
+  beforeEach(async () => {
+    resetDbChainMock()
+    root = await mkdtemp(join(tmpdir(), 'sim-project-prefix-'))
+    setUploadDirServer(root)
+  })
+
+  afterEach(async () => {
+    await rm(root, { recursive: true, force: true })
+  })
+
+  it('deletes the retired Project prefix and completes', async () => {
+    await put('project/p1/a.txt')
+    await put('project/p1/nested/b.txt')
+    await put('project/p10/keep.txt')
+
+    await expect(cleanup({ projectId: 'p1' }, outboxContext())).resolves.toBeUndefined()
+
+    expect(await exists('project/p1/a.txt')).toBe(false)
+    expect(await exists('project/p1/nested/b.txt')).toBe(false)
+    expect(await exists('project/p10/keep.txt')).toBe(true)
+  })
+
+  it('spends an attempt when a full page holds only objects it cannot remove', async () => {
+    for (let index = 0; index < 520; index++) await put(`project/p1/legacy file ${index}.txt`)
+
+    const result = await cleanup({ projectId: 'p1' }, outboxContext(Date.now() + 60_000))
+
+    expect(result).toMatchObject({ outcome: 'deferred' })
+    expect(result).not.toMatchObject({ consumeAttempt: false })
+  })
+})
diff --git a/apps/sim/lib/projects/files/prefix-cleanup.ts b/apps/sim/lib/projects/files/prefix-cleanup.ts
index 65b62c6b8d2..8f096a050a0 100644
--- a/apps/sim/lib/projects/files/prefix-cleanup.ts
+++ b/apps/sim/lib/projects/files/prefix-cleanup.ts
@@ -1,5 +1,3 @@
-import { opendir, rmdir } from 'node:fs/promises'
-import { dirname, join } from 'node:path'
 import { db } from '@sim/db'
 import { project, uploadSession } from '@sim/db/schema'
 import { describeError } from '@sim/utils/errors'
@@ -7,6 +5,7 @@ import { isRecordLike } from '@sim/utils/object'
 import { and, asc, eq, gt, inArray, isNull, sql } from 'drizzle-orm'
 import {
   continueOutboxHandler,
+  deferOutboxHandler,
   enqueueOutboxEvent,
   type OutboxEventContext,
   type OutboxHandler,
@@ -17,21 +16,19 @@ import {
   PROJECT_FILE_PREFIX_CLEANUP_EVENT,
   PROJECT_STORAGE_RECONCILE_EVENT,
 } from '@/lib/projects/files/outbox-events'
-import {
-  getStorageConfig,
-  USE_BLOB_STORAGE,
-  USE_GCS_STORAGE,
-  USE_S3_STORAGE,
-} from '@/lib/uploads/config'
 import { isObjectNotFoundError } from '@/lib/uploads/core/errors'
-import { UPLOAD_DIR_SERVER } from '@/lib/uploads/core/setup.server'
-import { deleteFile } from '@/lib/uploads/core/storage-service'
-import { abortProviderUpload, UPLOAD_URL_TTL_MS } from '@/lib/uploads/upload-session/provider'
+import {
+  deleteStorageKeysUnderPrefix,
+  listStorageKeyPage,
+  outboxDeadlineNear,
+  STORAGE_KEY_PAGE_SIZE,
+  storageCleanupAvailableAt,
+} from '@/lib/uploads/core/prefix-sweep'
+import { abortProviderUpload } from '@/lib/uploads/upload-session/provider'
 import type { UploadSessionRecord } from '@/lib/uploads/upload-session/service'
 import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/types'
 
 const PAGE_SIZE = 500
-const CLOCK_SKEW_MS = 60_000
 
 type UploadCleanup = Pick<
   UploadSessionRecord,
@@ -43,11 +40,6 @@ interface CleanupPayload {
   uploadCursor?: number
 }
 
-/** Expired signatures stop new transfers, but a transfer already in progress may finish later. */
-export function projectUploadCleanupAvailableAt(now = new Date()): Date {
-  return new Date(now.getTime() + UPLOAD_URL_TTL_MS + CLOCK_SKEW_MS)
-}
-
 function prefixFor(projectId: string): string {
   if (!/^[a-zA-Z0-9_-]+$/.test(projectId)) throw new Error('Invalid Project cleanup identity')
   return `project/${projectId}/`
@@ -101,7 +93,7 @@ async function enqueueSweeps(
 ) {
   await enqueueOutboxEvent(executor, PROJECT_FILE_PREFIX_CLEANUP_EVENT, payload)
   await enqueueOutboxEvent(executor, PROJECT_FILE_PREFIX_CLEANUP_EVENT, payload, {
-    availableAt: projectUploadCleanupAvailableAt(retiredAt),
+    availableAt: storageCleanupAvailableAt(retiredAt),
   })
 }
 
@@ -210,98 +202,21 @@ export async function queueRetiredProjectUploadCleanup(
   })
 }
 
-async function removeEmptyDirectory(path: string): Promise<void> {
-  try {
-    await rmdir(path)
-  } catch (error) {
-    if (!['ENOENT', 'ENOTEMPTY', 'EEXIST'].includes(describeError(error).code ?? '')) throw error
-  }
-}
-
-async function localKeys(prefix: string, signal: AbortSignal): Promise<string[]> {
-  const keys: string[] = []
-  async function visit(key: string): Promise<void> {
-    signal.throwIfAborted()
-    let directory
-    try {
-      directory = await opendir(join(UPLOAD_DIR_SERVER, key))
-    } catch (error) {
-      if (describeError(error).code === 'ENOENT') return
-      throw error
-    }
-    for await (const entry of directory) {
-      signal.throwIfAborted()
-      const child = `${key}${entry.name}`
-      if (entry.isDirectory()) await visit(`${child}/`)
-      else keys.push(child)
-      if (keys.length >= PAGE_SIZE) return
-    }
-    await removeEmptyDirectory(join(UPLOAD_DIR_SERVER, key))
-  }
-  await visit(prefix)
-  return keys
-}
-
-async function listKeys(prefix: string, signal: AbortSignal): Promise<string[]> {
-  signal.throwIfAborted()
-  if (USE_S3_STORAGE) {
-    const { ListObjectsV2Command } = await import('@aws-sdk/client-s3')
-    const { getS3Client } = await import('@/lib/uploads/providers/s3/client')
-    const config = getStorageConfig('project')
-    const page = await getS3Client().send(
-      new ListObjectsV2Command({ Bucket: config.bucket, Prefix: prefix, MaxKeys: PAGE_SIZE }),
-      { abortSignal: signal }
-    )
-    return (page.Contents ?? []).flatMap((object) => (object.Key ? [object.Key] : []))
-  }
-  if (USE_BLOB_STORAGE) {
-    const { getBlobServiceClient } = await import('@/lib/uploads/providers/blob/client')
-    const config = getStorageConfig('project')
-    if (!config.containerName) throw new Error('Project blob container missing')
-    const pages = (await getBlobServiceClient())
-      .getContainerClient(config.containerName)
-      .listBlobsFlat({ prefix, abortSignal: signal })
-      .byPage({ maxPageSize: PAGE_SIZE })
-    const page = await pages.next()
-    return page.done ? [] : page.value.segment.blobItems.map((object) => object.name)
-  }
-  if (USE_GCS_STORAGE) {
-    const { getGcsClient } = await import('@/lib/uploads/providers/google-cloud-storage/client')
-    const config = getStorageConfig('project')
-    if (!config.bucket) throw new Error('Project bucket missing')
-    const [files] = await (await getGcsClient())
-      .bucket(config.bucket)
-      .getFiles({ prefix, maxResults: PAGE_SIZE, autoPaginate: false })
-    signal.throwIfAborted()
-    return files.map((file) => file.name)
-  }
-  return localKeys(prefix, signal)
-}
-
-async function sweepPrefix(projectId: string, context: OutboxEventContext): Promise<boolean> {
+/** `stalled` when a page that must be re-listed removed nothing, so continuing cannot progress. */
+async function sweepPrefix(
+  projectId: string,
+  context: OutboxEventContext
+): Promise<'done' | 'continue' | 'stalled'> {
   const prefix = prefixFor(projectId)
-  const keys = await listKeys(prefix, context.signal)
-  let deleted = 0
-  for (const key of keys) {
-    if (!key.startsWith(prefix)) throw new Error('Storage provider returned a foreign Project key')
-    context.signal.throwIfAborted()
-    if (deleted > 0 && context.deadlineAt && Date.now() + 5000 >= context.deadlineAt) return false
-    try {
-      await deleteFile({ key, context: 'project', signal: context.signal })
-    } catch (error) {
-      if (describeError(error).code !== 'ENOENT') throw error
-    }
-    if (!USE_S3_STORAGE && !USE_BLOB_STORAGE && !USE_GCS_STORAGE) {
-      const root = join(UPLOAD_DIR_SERVER, 'project', projectId)
-      let directory = dirname(join(UPLOAD_DIR_SERVER, key))
-      while (directory === root || directory.startsWith(`${root}/`)) {
-        await removeEmptyDirectory(directory)
-        directory = dirname(directory)
-      }
-    }
-    deleted++
-  }
-  return keys.length < PAGE_SIZE
+  const page = await listStorageKeyPage(prefix, 'project', context.signal)
+  const { deleted, complete } = await deleteStorageKeysUnderPrefix(
+    prefix,
+    'project',
+    page.keys,
+    context
+  )
+  if (complete && page.keys.length < STORAGE_KEY_PAGE_SIZE) return 'done'
+  return deleted > 0 ? 'continue' : 'stalled'
 }
 
 const cleanupPrefix: OutboxHandler<unknown> = async (raw, context) => {
@@ -336,13 +251,15 @@ const cleanupPrefix: OutboxHandler<unknown> = async (raw, context) => {
     }
     if ((index + 1) % 20 === 0 || index === uploads.length - 1)
       await context.checkpointPayload({ uploadCursor: index + 1 })
-    if (context.deadlineAt && Date.now() + 5000 >= context.deadlineAt) {
+    if (outboxDeadlineNear(context)) {
       await context.checkpointPayload({ uploadCursor: index + 1 })
       return continueOutboxHandler('Continuing retired Project upload cleanup')
     }
   }
-  if (!(await sweepPrefix(payload.projectId, context)))
-    return continueOutboxHandler('Continuing Project storage cleanup')
+  const sweep = await sweepPrefix(payload.projectId, context)
+  if (sweep === 'continue') return continueOutboxHandler('Continuing Project storage cleanup')
+  if (sweep === 'stalled')
+    return deferOutboxHandler('Project storage cleanup removed nothing from a full listing page')
 }
 
 /**
diff --git a/apps/sim/lib/uploads/core/prefix-sweep.ts b/apps/sim/lib/uploads/core/prefix-sweep.ts
new file mode 100644
index 00000000000..40a183ee3dd
--- /dev/null
+++ b/apps/sim/lib/uploads/core/prefix-sweep.ts
@@ -0,0 +1,217 @@
+import { readdir, rmdir } from 'node:fs/promises'
+import { dirname, join } from 'node:path'
+import { describeError } from '@sim/utils/errors'
+import { isRecordLike } from '@sim/utils/object'
+import { compareStrings } from '@sim/utils/string'
+import type { OutboxEventContext } from '@/lib/core/outbox/service'
+import {
+  getStorageConfig,
+  type StorageContext,
+  USE_BLOB_STORAGE,
+  USE_GCS_STORAGE,
+  USE_S3_STORAGE,
+} from '@/lib/uploads/config'
+import { UPLOAD_DIR_SERVER } from '@/lib/uploads/core/setup.server'
+import { deleteFile } from '@/lib/uploads/core/storage-service'
+import { UPLOAD_URL_TTL_MS } from '@/lib/uploads/upload-session/provider'
+
+/** Most keys one {@link listStorageKeyPage} call returns. */
+export const STORAGE_KEY_PAGE_SIZE = 500
+const CLOCK_SKEW_MS = 60_000
+const DEADLINE_MARGIN_MS = 5000
+
+/**
+ * Earliest moment no upload URL signed before `now` can still write an object: expired signatures
+ * stop new transfers, but a transfer already in progress may finish later.
+ */
+export function storageCleanupAvailableAt(now = new Date()): Date {
+  return new Date(now.getTime() + UPLOAD_URL_TTL_MS + CLOCK_SKEW_MS)
+}
+
+/** True once an outbox handler should stop starting new storage work before its deadline. */
+export function outboxDeadlineNear(context: Pick<OutboxEventContext, 'deadlineAt'>): boolean {
+  return Boolean(context.deadlineAt && Date.now() + DEADLINE_MARGIN_MS >= context.deadlineAt)
+}
+
+async function removeEmptyDirectory(path: string): Promise<void> {
+  try {
+    await rmdir(path)
+  } catch (error) {
+    if (!['ENOENT', 'ENOTEMPTY', 'EEXIST'].includes(describeError(error).code ?? '')) throw error
+  }
+}
+
+/** Orders keys path segment by segment, the order a sorted depth-first walk visits them in. */
+function compareKeyPaths(left: readonly string[], right: readonly string[]): number {
+  for (let index = 0; index < Math.min(left.length, right.length); index++) {
+    const order = compareStrings(left[index], right[index])
+    if (order !== 0) return order
+  }
+  return left.length - right.length
+}
+
+/**
+ * The first local keys under `prefix` in {@link compareKeyPaths} order, after `startAfter` when
+ * given, so a caller can page past objects it keeps. Prunes directories it walks to the end empty.
+ */
+async function localKeys(
+  prefix: string,
+  signal: AbortSignal,
+  startAfter?: string
+): Promise<string[]> {
+  const cursor = startAfter?.split('/')
+  const keys: string[] = []
+  async function visit(key: string): Promise<void> {
+    signal.throwIfAborted()
+    let entries
+    try {
+      entries = await readdir(join(UPLOAD_DIR_SERVER, key), { withFileTypes: true })
+    } catch (error) {
+      if (describeError(error).code === 'ENOENT') return
+      throw error
+    }
+    entries.sort((left, right) => compareStrings(left.name, right.name))
+    for (const entry of entries) {
+      signal.throwIfAborted()
+      const child = `${key}${entry.name}`
+      if (entry.isDirectory()) {
+        const segments = child.split('/')
+        if (
+          !cursor ||
+          segments.every((segment, index) => cursor[index] === segment) ||
+          compareKeyPaths(segments, cursor) > 0
+        )
+          await visit(`${child}/`)
+      } else if (!cursor || compareKeyPaths(child.split('/'), cursor) > 0) {
+        keys.push(child)
+      }
+      if (keys.length >= STORAGE_KEY_PAGE_SIZE) return
+    }
+    await removeEmptyDirectory(join(UPLOAD_DIR_SERVER, key))
+  }
+  await visit(prefix)
+  return keys
+}
+
+/** One provider page of object keys; `done` once the listing has nothing after this page. */
+interface StorageKeyPage {
+  keys: string[]
+  done: boolean
+  /** Provider cursor for the following page, valid only within the current run. */
+  nextPageToken?: string
+}
+
+function keyPage(keys: string[], nextPageToken: string | undefined): StorageKeyPage {
+  return nextPageToken ? { keys, done: false, nextPageToken } : { keys, done: true }
+}
+
+/**
+ * Lists one page of object keys under `prefix` in the bucket serving `context`. A local listing's
+ * page token is the page's last key, so a caller keeping objects still advances past them.
+ */
+export async function listStorageKeyPage(
+  prefix: string,
+  context: StorageContext,
+  signal: AbortSignal,
+  pageToken?: string
+): Promise<StorageKeyPage> {
+  signal.throwIfAborted()
+  if (USE_S3_STORAGE) {
+    const { ListObjectsV2Command } = await import('@aws-sdk/client-s3')
+    const { getS3Client } = await import('@/lib/uploads/providers/s3/client')
+    const config = getStorageConfig(context)
+    const page = await getS3Client().send(
+      new ListObjectsV2Command({
+        Bucket: config.bucket,
+        Prefix: prefix,
+        MaxKeys: STORAGE_KEY_PAGE_SIZE,
+        ...(pageToken ? { ContinuationToken: pageToken } : {}),
+      }),
+      { abortSignal: signal }
+    )
+    return keyPage(
+      (page.Contents ?? []).flatMap((object) => (object.Key ? [object.Key] : [])),
+      page.IsTruncated ? page.NextContinuationToken : undefined
+    )
+  }
+  if (USE_BLOB_STORAGE) {
+    const { getBlobServiceClient } = await import('@/lib/uploads/providers/blob/client')
+    const config = getStorageConfig(context)
+    if (!config.containerName) throw new Error(`Blob container missing for ${context} storage`)
+    const pages = (await getBlobServiceClient())
+      .getContainerClient(config.containerName)
+      .listBlobsFlat({ prefix, abortSignal: signal })
+      .byPage({
+        maxPageSize: STORAGE_KEY_PAGE_SIZE,
+        ...(pageToken ? { continuationToken: pageToken } : {}),
+      })
+    const page = await pages.next()
+    if (page.done) return { keys: [], done: true }
+    return keyPage(
+      page.value.segment.blobItems.map((object) => object.name),
+      page.value.continuationToken || undefined
+    )
+  }
+  if (USE_GCS_STORAGE) {
+    const { getGcsClient } = await import('@/lib/uploads/providers/google-cloud-storage/client')
+    const config = getStorageConfig(context)
+    if (!config.bucket) throw new Error(`Bucket missing for ${context} storage`)
+    const [files, nextQuery] = await (await getGcsClient()).bucket(config.bucket).getFiles({
+      prefix,
+      maxResults: STORAGE_KEY_PAGE_SIZE,
+      autoPaginate: false,
+      ...(pageToken ? { pageToken } : {}),
+    })
+    signal.throwIfAborted()
+    return keyPage(
+      files.map((file) => file.name),
+      isRecordLike(nextQuery) && typeof nextQuery.pageToken === 'string'
+        ? nextQuery.pageToken || undefined
+        : undefined
+    )
+  }
+  const keys = await localKeys(prefix, signal, pageToken)
+  return keyPage(keys, keys.length < STORAGE_KEY_PAGE_SIZE ? undefined : keys[keys.length - 1])
+}
+
+/**
+ * Deletes listed keys, each of which must sit under `prefix`, and prunes local directories the
+ * deletions empty, never above the prefix itself. Stops before another deletion once the handler
+ * deadline is near; `complete` is false when it did. `deleted` counts only objects actually
+ * removed: a key already gone (or, locally, one whose on-disk name no storage key resolves to) is
+ * no progress, so a caller re-listing the same page can tell that it is stuck.
+ */
+export async function deleteStorageKeysUnderPrefix(
+  prefix: string,
+  context: StorageContext,
+  keys: readonly string[],
+  outbox: Pick<OutboxEventContext, 'signal' | 'deadlineAt'>
+): Promise<{ deleted: number; complete: boolean }> {
+  if (prefix.length < 2 || !prefix.endsWith('/')) throw new Error('Unbounded storage prefix')
+  let attempted = 0
+  let deleted = 0
+  for (const key of keys) {
+    if (!key.startsWith(prefix))
+      throw new Error('Storage provider returned a key outside the prefix')
+    outbox.signal.throwIfAborted()
+    if (attempted > 0 && outboxDeadlineNear(outbox)) return { deleted, complete: false }
+    attempted++
+    let removed = true
+    try {
+      await deleteFile({ key, context, signal: outbox.signal })
+    } catch (error) {
+      if (describeError(error).code !== 'ENOENT') throw error
+      removed = false
+    }
+    if (!USE_S3_STORAGE && !USE_BLOB_STORAGE && !USE_GCS_STORAGE) {
+      const root = join(UPLOAD_DIR_SERVER, prefix.slice(0, -1))
+      let directory = dirname(join(UPLOAD_DIR_SERVER, key))
+      while (directory === root || directory.startsWith(`${root}/`)) {
+        await removeEmptyDirectory(directory)
+        directory = dirname(directory)
+      }
+    }
+    if (removed) deleted++
+  }
+  return { deleted, complete: true }
+}
diff --git a/apps/sim/lib/uploads/shared/types.ts b/apps/sim/lib/uploads/shared/types.ts
index 45e5df6ad45..02502593d30 100644
--- a/apps/sim/lib/uploads/shared/types.ts
+++ b/apps/sim/lib/uploads/shared/types.ts
@@ -91,6 +91,19 @@ export type StorageContext =
  */
 export const WORKSPACE_SCOPED_CONTEXTS = ['workspace', 'mothership'] as const
 
+/**
+ * Contexts whose chat-less `workspace_files` rows belong to their workspace rather than their
+ * uploader: a departing member hands them to a successor. Private chat attachments keep their
+ * user's lifecycle even when stored under a workspace key.
+ */
+export const WORKSPACE_SHARED_FILE_CONTEXTS = [
+  'workspace',
+  'knowledge-base',
+  'execution',
+  'chat',
+  'workspace-logos',
+] as const
+
 export type WorkspaceScopedContext = (typeof WORKSPACE_SCOPED_CONTEXTS)[number]
 
 export function isWorkspaceScopedContext(
diff --git a/apps/sim/lib/uploads/utils/file.ts b/apps/sim/lib/uploads/utils/file.ts
index 43b15b67df9..97c373ba0ab 100644
--- a/apps/sim/lib/uploads/utils/file.ts
+++ b/apps/sim/lib/uploads/utils/file.ts
@@ -789,6 +789,12 @@ export function isPublicStorageContext(context: StorageContext): boolean {
   return PUBLIC_STORAGE_CONTEXTS.has(context)
 }
 
+/** Whether a stored key sits under a world-readable prefix, served without a row or a session. */
+export function isPublicStorageKey(key: string): boolean {
+  const context = tryInferContextFromKey(key)
+  return context !== null && isPublicStorageContext(context)
+}
+
 /**
  * Resolve the storage context for a stored file from its trusted key prefix.
  *
diff --git a/apps/sim/lib/users/account-deletion-attachments.test.ts b/apps/sim/lib/users/account-deletion-attachments.test.ts
index ff37d00302a..61f43e3fb99 100644
--- a/apps/sim/lib/users/account-deletion-attachments.test.ts
+++ b/apps/sim/lib/users/account-deletion-attachments.test.ts
@@ -90,6 +90,18 @@ describe('account deletion of private organization Assistant images', () => {
     }
   )
 
+  it('keeps an object a surviving row still references', async () => {
+    queueTableRows(schemaMock.uploadSession, [
+      { id: 'upload-1', key: IMAGE_KEY },
+      { id: 'upload-2', key: FAILED_IMAGE_KEY },
+    ])
+    queueTableRows(schemaMock.document, [{ key: IMAGE_KEY }])
+
+    await deleteUserAccount('user-1')
+
+    expect(mockDeleteFiles).toHaveBeenCalledExactlyOnceWith([FAILED_IMAGE_KEY], 'mothership')
+  })
+
   it('retains ownership while an issued upload URL could recreate a purged object', async () => {
     queueTableRows(schemaMock.uploadSession, [{ id: 'upload-1', key: IMAGE_KEY }])
 
diff --git a/apps/sim/lib/users/account-deletion.test.ts b/apps/sim/lib/users/account-deletion.test.ts
index 1991d10b78a..1b67261fbb1 100644
--- a/apps/sim/lib/users/account-deletion.test.ts
+++ b/apps/sim/lib/users/account-deletion.test.ts
@@ -2,10 +2,10 @@ import { describe, expect, it } from 'vitest'
 import {
   type AccountDeletionFacts,
   classifyAccountDeletion,
-  extractProfilePictureKey,
   type WorkspaceCompany,
   type WorkspaceRow,
 } from '@/lib/users/account-deletion'
+import { extractProfilePictureKey } from '@/lib/users/storage-sweep'
 
 function workspace(overrides: Partial<WorkspaceRow> = {}): WorkspaceRow {
   return { id: 'ws-1', name: 'My workspace', organizationId: null, ...overrides }
diff --git a/apps/sim/lib/users/account-deletion.ts b/apps/sim/lib/users/account-deletion.ts
index ca0c883f95f..89c70597874 100644
--- a/apps/sim/lib/users/account-deletion.ts
+++ b/apps/sim/lib/users/account-deletion.ts
@@ -54,6 +54,12 @@ import {
 } from '@/lib/table/rows/executions'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
+import { isPublicStorageKey } from '@/lib/uploads/utils/file'
+import { createStorageReferenceGuard } from '@/lib/users/storage-reference-guard'
+import {
+  collectAccountStorageObjects,
+  enqueueAccountStorageCleanupInTx,
+} from '@/lib/users/storage-sweep'
 import {
   listSharedResourceWorkspaceIdsForUser,
   listUnresolvedSharedResourceWorkspaces,
@@ -492,31 +498,6 @@ function countKeys(batches: StorageKeyBatch[]): number {
   return total
 }
 
-/**
- * The account's own uploaded avatar, as a storage key.
- *
- * `user.image` holds either a `/api/files/serve/...` path (an upload we own) or
- * an absolute URL from an OAuth provider (which we must not try to delete). Only
- * the former yields a key, and only under the `profile-pictures/` prefix — the
- * image is personal data, so an erasure that leaves it in the bucket is not an
- * erasure.
- */
-export function extractProfilePictureKey(image: string | null): string | null {
-  if (!image) return null
-  try {
-    const parsed = new URL(image, 'http://placeholder')
-    if (parsed.origin !== 'http://placeholder') return null
-    const segments = parsed.pathname.split('/')
-    if (segments[1] !== 'api' || segments[2] !== 'files' || segments[3] !== 'serve') return null
-    let keySegments = segments.slice(4)
-    if (['s3', 'blob', 'gcs'].includes(keySegments[0])) keySegments = keySegments.slice(1)
-    const key = decodeURIComponent(keySegments.join('/'))
-    return key.startsWith('profile-pictures/') ? key : null
-  } catch {
-    return null
-  }
-}
-
 /**
  * Collects the account's private images and stored objects in workspaces that go with it.
  *
@@ -555,16 +536,6 @@ async function collectAccountStorageKeys(
 
   if (!isUsingCloudStorage()) return batches
 
-  const [profile] = await db
-    .select({ image: user.image })
-    .from(user)
-    .where(eq(user.id, userId))
-    .limit(1)
-  const profilePictureKey = extractProfilePictureKey(profile?.image ?? null)
-  if (profilePictureKey) {
-    batches.push({ context: 'profile-pictures', keys: [profilePictureKey] })
-  }
-
   if (workspaceIds.length === 0) return batches
 
   await collectPages(
@@ -637,11 +608,27 @@ async function collectAccountStorageKeys(
  * time this runs the account is already gone, and the request must not report a
  * failure for work that cannot be undone. An orphaned object is recoverable from
  * the log; a deletion the caller believes failed is not.
+ *
+ * Keys a surviving row still names are kept (a legacy `kb/` object can back
+ * documents in more than one workspace); see `createStorageReferenceGuard`.
+ * World-readable keys are left to the delayed durable cleanup, whose guard also
+ * reads stored asset URLs, so this request never does.
  */
 async function purgeStorageObjects(userId: string, batches: StorageKeyBatch[]): Promise<void> {
-  for (const { context, keys } of batches) {
-    if (keys.length === 0) continue
+  const isReferenced = createStorageReferenceGuard()
+  for (const { context, keys: listed } of batches) {
+    const collected = listed.filter((key) => !isPublicStorageKey(key))
+    if (collected.length === 0) continue
     try {
+      const referenced = await isReferenced(collected)
+      if (referenced.size > 0) {
+        logger.warn('Keeping objects a surviving reference still names', {
+          context,
+          kept: referenced.size,
+        })
+      }
+      const keys = collected.filter((key) => !referenced.has(key))
+      if (keys.length === 0) continue
       const { failed } = await StorageService.deleteFiles(keys, context)
       for (const { key, error } of failed) {
         logger.error('Failed to erase stored object during account deletion', {
@@ -778,6 +765,10 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
   const doomedWorkspaceIds = plan.workspacesToDelete.map((workspace) => workspace.id)
   const doomed = new Set(doomedWorkspaceIds)
   const storageKeys = await collectAccountStorageKeys(userId, doomedWorkspaceIds)
+  const durableStorageObjects = await collectAccountStorageObjects({
+    userId,
+    workspaceIds: doomedWorkspaceIds,
+  })
 
   let cancelledDispatches: CancelledDispatch[] = []
   let cancelledMarkers: CancelledCellMarker[] = []
@@ -861,6 +852,12 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
      * workspace and user deletes cascade their rows away; delivered after commit.
      */
     stoppedRuns = await stopLiveRunsTx(tx, { userId })
+    /** Queues the pre-read keys plus a delta of the account's rows written since, read under the account lock. */
+    await enqueueAccountStorageCleanupInTx(tx, {
+      userId,
+      workspaceIds: doomedWorkspaceIds,
+      objects: durableStorageObjects,
+    })
     if (doomedWorkspaceIds.length > 0) {
       /**
        * Re-checked here rather than trusted from the plan: a workspace that
diff --git a/apps/sim/lib/users/storage-reference-guard.ts b/apps/sim/lib/users/storage-reference-guard.ts
new file mode 100644
index 00000000000..bf55b46af8f
--- /dev/null
+++ b/apps/sim/lib/users/storage-reference-guard.ts
@@ -0,0 +1,269 @@
+import { db } from '@sim/db'
+import {
+  chat,
+  customBlock,
+  document,
+  executionLargeValues,
+  oauthClient,
+  organization,
+  user,
+  workspaceFile,
+  workspaceFiles,
+  workspaceFileVersion,
+  workspace as workspaceTable,
+} from '@sim/db/schema'
+import { chunkArray } from '@sim/utils/helpers'
+import { isRecordLike } from '@sim/utils/object'
+import { and, gt, inArray, or, type SQL, type SQLWrapper, sql } from 'drizzle-orm'
+import { extractStorageKey, isPublicStorageKey } from '@/lib/uploads/utils/file'
+import { keysUnderLiveWorkspaces } from '@/lib/users/workspace-storage'
+
+/** Every string inside a URL column or a JSON settings blob, including JSON held as a string. */
+function collectStrings(value: unknown, into: string[]): void {
+  if (typeof value === 'string') {
+    into.push(value)
+    const trimmed = value.trim()
+    if (trimmed.startsWith('{') || trimmed.startsWith('[')) {
+      try {
+        collectStrings(JSON.parse(trimmed), into)
+      } catch {}
+    }
+  } else if (Array.isArray(value)) {
+    for (const item of value) collectStrings(item, into)
+  } else if (isRecordLike(value)) {
+    for (const item of Object.values(value)) collectStrings(item, into)
+  }
+}
+
+const SERVE_PATH = '/api/files/serve/'
+
+/**
+ * The storage keys a stored URL can name: a `/api/files/serve/...` path (relative or absolute,
+ * with or without a provider segment), or a direct bucket URL in virtual-host or path style.
+ * Over-matching only keeps an object. Never throws: a value whose percent-encoding is malformed
+ * yields its raw path and that path with each valid escape decoded (a legacy key can hold a
+ * literal `%`), so one bad stored string cannot disable every purge.
+ */
+function keysNamedByUrl(value: string): string[] {
+  const keys = new Set<string>()
+  const add = (candidate: string) => {
+    if (!candidate) return
+    keys.add(candidate)
+    try {
+      keys.add(decodeURIComponent(candidate))
+    } catch {
+      keys.add(
+        candidate.replace(/(%[0-9A-Fa-f]{2})+/g, (escapes) => {
+          try {
+            return decodeURIComponent(escapes)
+          } catch {
+            return escapes
+          }
+        })
+      )
+    }
+  }
+  const trimmed = value.trim()
+  let path: string
+  try {
+    path = extractStorageKey(trimmed)
+  } catch {
+    const raw = trimmed.split('?')[0]
+    const serveAt = raw.indexOf(SERVE_PATH)
+    path = serveAt >= 0 ? raw.slice(serveAt + SERVE_PATH.length) : raw
+  }
+  path = path.replace(/^\/+/, '')
+  add(path)
+  const slash = path.indexOf('/')
+  if (slash > 0) add(path.slice(slash + 1))
+  return [...keys]
+}
+
+const MIN_OWNER_RUN_LENGTH = 8
+const OWNER_RUNS_PER_QUERY = 50
+const URL_ROW_PAGE_SIZE = 500
+
+/**
+ * The run of `[A-Za-z0-9_-]` that opens the segment after a public key's root: the owning
+ * workspace or organization id (`workspace-logos/<id>/...`), or the upload id a flat key starts
+ * with (`profile-pictures/<id>-<name>`). `encodeURIComponent` and `encodeURI` never escape those
+ * characters, so every URL this app writes for the key holds the run literally. `null` when the
+ * run is too short to narrow a lookup.
+ */
+function ownerRun(key: string): string | null {
+  const run = /^[A-Za-z0-9_-]+/.exec(key.split('/')[1] ?? '')?.[0] ?? ''
+  return run.length >= MIN_OWNER_RUN_LENGTH ? run : null
+}
+
+type UrlRowPage = (
+  mentions: (column: SQLWrapper) => SQL,
+  afterId: string
+) => Promise<{ id: string; values: unknown[] }[]>
+
+/**
+ * The columns that hold a stored asset as a URL rather than a row: `user.image`,
+ * `workspace.logo_url`, `custom_block.icon_url`, `chat.customizations` (deployed-chat logo and
+ * avatar), `oauth_client.icon`, and `organization.logo`, `.whitelabel_settings`, `.metadata`.
+ */
+const URL_ROW_PAGES: readonly UrlRowPage[] = [
+  (mentions, afterId) =>
+    db
+      .select({ id: user.id, value: user.image })
+      .from(user)
+      .where(and(mentions(user.image), gt(user.id, afterId)))
+      .orderBy(user.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) => rows.map((row) => ({ id: row.id, values: [row.value] }))),
+  (mentions, afterId) =>
+    db
+      .select({ id: workspaceTable.id, value: workspaceTable.logoUrl })
+      .from(workspaceTable)
+      .where(and(mentions(workspaceTable.logoUrl), gt(workspaceTable.id, afterId)))
+      .orderBy(workspaceTable.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) => rows.map((row) => ({ id: row.id, values: [row.value] }))),
+  (mentions, afterId) =>
+    db
+      .select({ id: customBlock.id, value: customBlock.iconUrl })
+      .from(customBlock)
+      .where(and(mentions(customBlock.iconUrl), gt(customBlock.id, afterId)))
+      .orderBy(customBlock.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) => rows.map((row) => ({ id: row.id, values: [row.value] }))),
+  (mentions, afterId) =>
+    db
+      .select({ id: chat.id, value: chat.customizations })
+      .from(chat)
+      .where(and(mentions(chat.customizations), gt(chat.id, afterId)))
+      .orderBy(chat.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) => rows.map((row) => ({ id: row.id, values: [row.value] }))),
+  (mentions, afterId) =>
+    db
+      .select({ id: oauthClient.id, value: oauthClient.icon })
+      .from(oauthClient)
+      .where(and(mentions(oauthClient.icon), gt(oauthClient.id, afterId)))
+      .orderBy(oauthClient.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) => rows.map((row) => ({ id: row.id, values: [row.value] }))),
+  (mentions, afterId) =>
+    db
+      .select({
+        id: organization.id,
+        logo: organization.logo,
+        whitelabel: organization.whitelabelSettings,
+        metadata: organization.metadata,
+      })
+      .from(organization)
+      .where(
+        and(
+          or(
+            mentions(organization.logo),
+            mentions(organization.whitelabelSettings),
+            mentions(organization.metadata)
+          ),
+          gt(organization.id, afterId)
+        )
+      )
+      .orderBy(organization.id)
+      .limit(URL_ROW_PAGE_SIZE)
+      .then((rows) =>
+        rows.map((row) => ({ id: row.id, values: [row.logo, row.whitelabel, row.metadata] }))
+      ),
+]
+
+/**
+ * Adds to `named` every key a stored asset URL names, reading only rows whose URL columns mention
+ * one of `runs` (see {@link ownerRun}), page by page.
+ */
+async function loadKeysNamedByUrls(runs: readonly string[], named: Set<string>): Promise<void> {
+  const mentions = (column: SQLWrapper): SQL =>
+    sql`(${sql.join(
+      runs.map((run) => sql`${column}::text LIKE ${`%${run}%`}`),
+      sql` OR `
+    )})`
+  for (const page of URL_ROW_PAGES) {
+    let afterId = ''
+    for (;;) {
+      const rows = await page(mentions, afterId)
+      for (const row of rows) {
+        const values: string[] = []
+        collectStrings(row.values, values)
+        for (const value of values) for (const key of keysNamedByUrl(value)) named.add(key)
+      }
+      if (rows.length < URL_ROW_PAGE_SIZE) break
+      afterId = rows[rows.length - 1].id
+    }
+  }
+}
+
+/**
+ * Builds the check every account-deletion purge runs before deleting a key. A key is kept when:
+ *
+ * - a row still names it: `workspace_files.key`, `workspace_file.key`,
+ *   `workspace_file_version.key`, `document.storage_key`, `execution_large_values.key`;
+ * - it sits under a workspace-keyed root of a workspace that still exists
+ *   ({@link keysUnderLiveWorkspaces}); or
+ * - it sits under a world-readable root (`isPublicStorageKey`, the same prefixes the file-serve
+ *   route's `isPublicByKeyPrefix` answers without a session) and a stored asset URL names it.
+ *   Only URL rows that mention the key's {@link ownerRun} are read, on every call so a reference
+ *   written mid-sweep still counts; a public key without one is kept unread. A URL that
+ *   hand-encodes characters of that run is missed. Account deletion asks about public keys only
+ *   under a deleted workspace's prefix (its logos) and for the avatar the erased account's
+ *   `user.image` names, so such a URL can cost one of those, and only when no surviving reference
+ *   names it the way this app writes URLs; another user's own `user.image` naming their avatar
+ *   always keeps it.
+ *
+ * No URL column is read for any other key: `verifyFileAccess` serves it only through a row, by its
+ * key path under a workspace that still exists (kept above), or by object metadata to a signed-in holder of a hand-pasted URL to
+ * a private file this deletion means to erase.
+ *
+ * Every row of a deleted workspace or account is gone when this runs, so a remaining reference
+ * belongs to data that outlived the deletion — a moved knowledge base, a handed-off file, an
+ * organization logo. References embedded in free-form JSON or text — workflow sub-block values,
+ * chat messages, Markdown links — are not consulted; for the deleted account's own chat
+ * attachments that is the intended erasure. A failed lookup throws, so nothing is deleted.
+ */
+export function createStorageReferenceGuard(): (keys: readonly string[]) => Promise<Set<string>> {
+  return async (keys) => {
+    const referenced = new Set<string>()
+    if (keys.length === 0) return referenced
+    const list = [...new Set(keys)]
+    const rows = [
+      ...(await db
+        .select({ key: workspaceFiles.key })
+        .from(workspaceFiles)
+        .where(inArray(workspaceFiles.key, list))),
+      ...(await db
+        .select({ key: workspaceFile.key })
+        .from(workspaceFile)
+        .where(inArray(workspaceFile.key, list))),
+      ...(await db
+        .select({ key: workspaceFileVersion.key })
+        .from(workspaceFileVersion)
+        .where(inArray(workspaceFileVersion.key, list))),
+      ...(await db
+        .select({ key: document.storageKey })
+        .from(document)
+        .where(inArray(document.storageKey, list))),
+      ...(await db
+        .select({ key: executionLargeValues.key })
+        .from(executionLargeValues)
+        .where(inArray(executionLargeValues.key, list))),
+    ]
+    for (const row of rows) if (row.key) referenced.add(row.key)
+    const publicKeys = list.filter(isPublicStorageKey)
+    const runs = new Set<string>()
+    for (const key of publicKeys) {
+      const run = ownerRun(key)
+      if (run) runs.add(run)
+      else referenced.add(key)
+    }
+    const named = new Set<string>()
+    for (const batch of chunkArray([...runs], OWNER_RUNS_PER_QUERY))
+      await loadKeysNamedByUrls(batch, named)
+    for (const key of publicKeys) if (named.has(key)) referenced.add(key)
+    for (const key of await keysUnderLiveWorkspaces(list)) referenced.add(key)
+    return referenced
+  }
+}
diff --git a/apps/sim/lib/users/storage-sweep-events.ts b/apps/sim/lib/users/storage-sweep-events.ts
new file mode 100644
index 00000000000..b709df01425
--- /dev/null
+++ b/apps/sim/lib/users/storage-sweep-events.ts
@@ -0,0 +1,2 @@
+export const ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT = 'account-deletion.storage.workspace-sweep'
+export const ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT = 'account-deletion.storage.object-cleanup'
diff --git a/apps/sim/lib/users/storage-sweep.integration.ts b/apps/sim/lib/users/storage-sweep.integration.ts
new file mode 100644
index 00000000000..6de24936293
--- /dev/null
+++ b/apps/sim/lib/users/storage-sweep.integration.ts
@@ -0,0 +1,384 @@
+import { mkdtempSync } from 'node:fs'
+import { mkdir, readFile, rm, stat, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { dirname, join } from 'node:path'
+import { db } from '@sim/db'
+import {
+  copilotChats,
+  outboxEvent,
+  permissions,
+  user,
+  workspace,
+  workspaceFiles,
+} from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
+import { setUploadDirServer, uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock'
+import { getErrorMessage } from '@sim/utils/errors'
+import { generateId } from '@sim/utils/id'
+import { and, eq, inArray, or, sql } from 'drizzle-orm'
+import { afterAll, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock)
+
+import { processOutboxEventById } from '@/lib/core/outbox/service'
+import { UPLOAD_URL_TTL_MS } from '@/lib/uploads/upload-session/provider'
+import { deleteUserAccount } from '@/lib/users/account-deletion'
+import { accountStorageSweepOutboxHandlers } from '@/lib/users/storage-sweep'
+import {
+  ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT,
+  ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+} from '@/lib/users/storage-sweep-events'
+
+readTestDatabaseUrl()
+
+const storageRoot = mkdtempSync(join(tmpdir(), 'sim-account-storage-sweep-'))
+setUploadDirServer(storageRoot)
+
+const userIds: string[] = []
+const workspaceIds: string[] = []
+const checks: { name: string; status: 'passed' | 'failed'; durationMs: number; error?: string }[] =
+  []
+
+function check(name: string, run: () => Promise<void>) {
+  it(name, async () => {
+    const started = performance.now()
+    try {
+      await run()
+      checks.push({ name, status: 'passed', durationMs: performance.now() - started })
+    } catch (error) {
+      checks.push({
+        name,
+        status: 'failed',
+        durationMs: performance.now() - started,
+        error: getErrorMessage(error),
+      })
+      throw error
+    }
+  })
+}
+
+async function put(key: string, body = 'x'): Promise<void> {
+  await mkdir(dirname(join(storageRoot, key)), { recursive: true })
+  await writeFile(join(storageRoot, key), body)
+}
+
+async function exists(key: string): Promise<boolean> {
+  try {
+    await stat(join(storageRoot, key))
+    return true
+  } catch {
+    return false
+  }
+}
+
+async function insertUsers(ids: string[]) {
+  userIds.push(...ids)
+  await db.insert(user).values(
+    ids.map((id) => ({
+      id,
+      name: 'Storage sweep fixture',
+      email: `${id}@sweep.invalid`,
+      emailVerified: true,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+    }))
+  )
+}
+
+async function insertAdminWorkspace(id: string, ownerId: string) {
+  workspaceIds.push(id)
+  await insertWorkspaceFixture(db, {
+    id,
+    name: 'Storage sweep fixture',
+    ownerId,
+    billedAccountUserId: ownerId,
+  })
+  await db.insert(permissions).values({
+    id: generateId(),
+    userId: ownerId,
+    entityType: 'workspace',
+    entityId: id,
+    permissionType: 'admin',
+  })
+}
+
+function avatarUrl(key: string): string {
+  return `/api/files/serve/${encodeURIComponent(key)}?context=profile-pictures`
+}
+
+function fileRow(values: {
+  key: string
+  userId: string
+  workspaceId: string | null
+  context: string
+  chatId?: string
+}) {
+  return {
+    id: generateId(),
+    originalName: `fixture-${generateId()}.txt`,
+    contentType: 'text/plain',
+    sizeBytes: 1,
+    chatId: values.chatId ?? null,
+    ...values,
+  }
+}
+
+/**
+ * A departing account with a private workspace, and files it left in another person's workspace
+ * after departing it: one handed to the workspace's billing account, one private chat attachment,
+ * one attached to the other person's chat, and one with no workspace at all.
+ */
+async function seed() {
+  const ownerId = generateId()
+  const departingId = generateId()
+  const doomedId = generateId()
+  const survivingId = generateId()
+  await insertUsers([ownerId, departingId])
+  await insertAdminWorkspace(doomedId, departingId)
+  await insertAdminWorkspace(survivingId, ownerId)
+  const departingChatId = generateId()
+  const ownerChatId = generateId()
+  await db.insert(copilotChats).values([
+    { id: departingChatId, userId: departingId, workspaceId: survivingId },
+    { id: ownerChatId, userId: ownerId, workspaceId: survivingId },
+  ])
+  const keys = {
+    doomedFile: `workspace/${doomedId}/${generateId()}.txt`,
+    doomedCopilot: `copilot/${generateId()}/doomed.txt`,
+    rowlessExecution: `execution/${doomedId}/wf/ex/large-value-lv_rowless.json`,
+    rowlessSnapshot: `table-snapshots/${doomedId}/table/v1-hash.csv`,
+    rowlessCompiled: `copilot-doc-compiled/${doomedId}/hash.pdf`,
+    referencedElsewhere: `workspace/${doomedId}/moved-${generateId()}.txt`,
+    neighbor: `workspace/${doomedId}x/keep.txt`,
+    handedOff: `workspace/${survivingId}/${generateId()}.txt`,
+    privateAttachment: `workspace/${survivingId}/${generateId()}/private.txt`,
+    othersChatAttachment: `workspace/${survivingId}/${generateId()}/others.txt`,
+    personalCopilot: `copilot/${generateId()}/personal.txt`,
+    legacyWorkspacelessKb: `kb/${generateId()}-legacy.txt`,
+    profilePictureRowLogo: `profile-pictures/${generateId()}/org-logo.png`,
+    logoNamedByUrl: `workspace-logos/${doomedId}/${generateId()}-org-logo.png`,
+    logoNamedByEncodedUrl: `workspace-logos/${doomedId}/${generateId()}-workspace-logo.png`,
+    ownAvatar: `profile-pictures/${generateId()}-departing-avatar.png`,
+    othersAvatar: `profile-pictures/${generateId()}-owner-avatar.png`,
+  }
+  await db.insert(workspaceFiles).values([
+    fileRow({
+      key: keys.doomedFile,
+      userId: departingId,
+      workspaceId: doomedId,
+      context: 'workspace',
+    }),
+    fileRow({
+      key: keys.doomedCopilot,
+      userId: departingId,
+      workspaceId: doomedId,
+      context: 'copilot',
+    }),
+    fileRow({
+      key: keys.referencedElsewhere,
+      userId: ownerId,
+      workspaceId: survivingId,
+      context: 'workspace',
+    }),
+    fileRow({
+      key: keys.handedOff,
+      userId: departingId,
+      workspaceId: survivingId,
+      context: 'workspace',
+    }),
+    fileRow({
+      key: keys.privateAttachment,
+      userId: departingId,
+      workspaceId: survivingId,
+      context: 'mothership',
+      chatId: departingChatId,
+    }),
+    fileRow({
+      key: keys.othersChatAttachment,
+      userId: departingId,
+      workspaceId: survivingId,
+      context: 'mothership',
+      chatId: ownerChatId,
+    }),
+    fileRow({
+      key: keys.personalCopilot,
+      userId: departingId,
+      workspaceId: null,
+      context: 'copilot',
+    }),
+    fileRow({
+      key: keys.legacyWorkspacelessKb,
+      userId: departingId,
+      workspaceId: null,
+      context: 'knowledge-base',
+    }),
+    fileRow({
+      key: keys.profilePictureRowLogo,
+      userId: departingId,
+      workspaceId: null,
+      context: 'profile-pictures',
+    }),
+  ])
+  await db
+    .update(user)
+    .set({ image: `/api/files/serve/${encodeURIComponent(keys.logoNamedByUrl)}` })
+    .where(eq(user.id, ownerId))
+  await db
+    .update(workspace)
+    .set({
+      logoUrl: `/api/files/serve/s3/${encodeURIComponent(keys.logoNamedByEncodedUrl)}`,
+    })
+    .where(eq(workspace.id, survivingId))
+  await db
+    .update(user)
+    .set({ image: avatarUrl(keys.ownAvatar) })
+    .where(eq(user.id, departingId))
+  for (const key of Object.values(keys)) await put(key)
+  return { ownerId, departingId, doomedId, survivingId, keys }
+}
+
+function sweepEvents(workspaceId: string) {
+  return db
+    .select()
+    .from(outboxEvent)
+    .where(
+      and(
+        eq(outboxEvent.eventType, ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT),
+        sql`${outboxEvent.payload}->>'workspaceId' = ${workspaceId}`
+      )
+    )
+}
+
+function objectEvents(userId: string) {
+  return db
+    .select()
+    .from(outboxEvent)
+    .where(
+      and(
+        eq(outboxEvent.eventType, ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT),
+        sql`${outboxEvent.payload}->>'userId' = ${userId}`
+      )
+    )
+}
+
+async function runNow(eventIds: string[]) {
+  for (const id of eventIds) {
+    await db
+      .update(outboxEvent)
+      .set({ availableAt: new Date(0) })
+      .where(eq(outboxEvent.id, id))
+    expect(await processOutboxEventById(id, accountStorageSweepOutboxHandlers)).toBe('completed')
+  }
+}
+
+describe('account deletion storage sweep', () => {
+  check(
+    'purges the deleted workspace by prefix and the account private files by key, keeping every live object',
+    async () => {
+      const f = await seed()
+      const startedAt = Date.now()
+      await deleteUserAccount(f.departingId)
+
+      const sweeps = await sweepEvents(f.doomedId)
+      const objects = await objectEvents(f.departingId)
+      expect(sweeps).toHaveLength(1)
+      expect(objects.length).toBeGreaterThan(0)
+      for (const event of [...sweeps, ...objects]) {
+        expect(event.availableAt.getTime()).toBeGreaterThanOrEqual(startedAt + UPLOAD_URL_TTL_MS)
+        expect(await processOutboxEventById(event.id, accountStorageSweepOutboxHandlers)).toBe(
+          'pending'
+        )
+      }
+      const queuedKeys = objects.flatMap(
+        (event) => (event.payload as { objects: { key: string }[] }).objects
+      )
+      expect(queuedKeys.map((object) => object.key)).not.toContain(f.keys.othersChatAttachment)
+      expect(queuedKeys.map((object) => object.key)).not.toContain(f.keys.handedOff)
+      expect(queuedKeys.map((object) => object.key)).not.toContain(f.keys.legacyWorkspacelessKb)
+      expect(queuedKeys.map((object) => object.key)).not.toContain(f.keys.profilePictureRowLogo)
+
+      const lateUpload = `workspace/${f.doomedId}/late-${generateId()}.bin`
+      await put(lateUpload)
+      await runNow([...sweeps, ...objects].map((event) => event.id))
+
+      for (const key of [
+        f.keys.doomedFile,
+        f.keys.doomedCopilot,
+        f.keys.rowlessExecution,
+        f.keys.rowlessSnapshot,
+        f.keys.rowlessCompiled,
+        f.keys.personalCopilot,
+        f.keys.ownAvatar,
+        lateUpload,
+      ])
+        expect({ key, exists: await exists(key) }).toEqual({ key, exists: false })
+      for (const key of [
+        f.keys.referencedElsewhere,
+        f.keys.neighbor,
+        f.keys.handedOff,
+        f.keys.othersChatAttachment,
+        f.keys.privateAttachment,
+        f.keys.legacyWorkspacelessKb,
+        f.keys.profilePictureRowLogo,
+        f.keys.logoNamedByUrl,
+        f.keys.logoNamedByEncodedUrl,
+        f.keys.othersAvatar,
+      ])
+        expect(await readFile(join(storageRoot, key), 'utf8')).toBe('x')
+      expect(
+        await db.select().from(workspaceFiles).where(eq(workspaceFiles.key, f.keys.handedOff))
+      ).toMatchObject([{ userId: f.ownerId }])
+    }
+  )
+
+  check("keeps another account's avatar the departing image points at", async () => {
+    const f = await seed()
+    await db
+      .update(user)
+      .set({ image: avatarUrl(f.keys.othersAvatar) })
+      .where(inArray(user.id, [f.ownerId, f.departingId]))
+    await deleteUserAccount(f.departingId)
+
+    await runNow((await objectEvents(f.departingId)).map((event) => event.id))
+
+    expect(await readFile(join(storageRoot, f.keys.othersAvatar), 'utf8')).toBe('x')
+  })
+
+  check('a sweep for a workspace that exists again completes without deleting', async () => {
+    const f = await seed()
+    await deleteUserAccount(f.departingId)
+    const [sweep] = await sweepEvents(f.doomedId)
+    await insertAdminWorkspace(f.doomedId, f.ownerId)
+    const restored = `workspace/${f.doomedId}/restored.txt`
+    await put(restored)
+
+    await runNow([sweep.id])
+
+    expect(await readFile(join(storageRoot, restored), 'utf8')).toBe('x')
+    expect(await readFile(join(storageRoot, f.keys.rowlessExecution), 'utf8')).toBe('x')
+  })
+})
+
+afterAll(async () => {
+  try {
+    await db
+      .delete(outboxEvent)
+      .where(
+        or(
+          inArray(sql`${outboxEvent.payload}->>'workspaceId'`, workspaceIds),
+          inArray(sql`${outboxEvent.payload}->>'userId'`, userIds)
+        )
+      )
+    await db.delete(workspaceFiles).where(inArray(workspaceFiles.userId, userIds))
+    await deleteWorkspaceFixture(db, inArray(workspace.id, workspaceIds))
+    await db.delete(user).where(inArray(user.id, userIds))
+    await rm(storageRoot, { recursive: true, force: true })
+  } finally {
+    const reportPath = process.env.ACCOUNT_STORAGE_SWEEP_REPORT_PATH
+    if (reportPath) {
+      await mkdir(dirname(reportPath), { recursive: true })
+      await writeFile(reportPath, JSON.stringify({ checks }, null, 2))
+    }
+  }
+})
diff --git a/apps/sim/lib/users/storage-sweep.test.ts b/apps/sim/lib/users/storage-sweep.test.ts
new file mode 100644
index 00000000000..c9cb20e1d88
--- /dev/null
+++ b/apps/sim/lib/users/storage-sweep.test.ts
@@ -0,0 +1,547 @@
+import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { dirname, join } from 'node:path'
+import {
+  dbChainMock,
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  schemaMock,
+} from '@sim/testing'
+import { uploadsConfigMock } from '@sim/testing/mocks/uploads-config.mock'
+import { uploadsMetadataMock } from '@sim/testing/mocks/uploads-metadata.mock'
+import { setUploadDirServer, uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock)
+vi.mock('@/lib/uploads/config', () => uploadsConfigMock)
+vi.mock('@/lib/uploads/server/metadata', () => uploadsMetadataMock)
+
+import type { OutboxEventContext } from '@/lib/core/outbox/service'
+import type { DbTransaction } from '@/lib/db/types'
+import { createStorageReferenceGuard } from '@/lib/users/storage-reference-guard'
+import {
+  accountStorageSweepOutboxHandlers,
+  collectAccountStorageObjects,
+  enqueueAccountStorageCleanupInTx,
+} from '@/lib/users/storage-sweep'
+import {
+  ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT,
+  ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+} from '@/lib/users/storage-sweep-events'
+
+const sweep = accountStorageSweepOutboxHandlers[ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT]
+const cleanup = accountStorageSweepOutboxHandlers[ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT]
+
+let root: string
+
+function outboxContext(overrides: Partial<OutboxEventContext> = {}): OutboxEventContext {
+  return {
+    eventId: 'event-1',
+    eventType: ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+    attempts: 0,
+    maxAttempts: 10,
+    signal: new AbortController().signal,
+    checkpointPayload: vi.fn(async () => {}),
+    ...overrides,
+  }
+}
+
+async function put(key: string, body = 'x'): Promise<void> {
+  await mkdir(dirname(join(root, key)), { recursive: true })
+  await writeFile(join(root, key), body)
+}
+
+async function exists(key: string): Promise<boolean> {
+  try {
+    await stat(join(root, key))
+    return true
+  } catch {
+    return false
+  }
+}
+
+const DOOMED_KEYS = [
+  'workspace/wsdoomed1/a.txt',
+  'workspace/wsdoomed1/nested/b.txt',
+  'table-import/wsdoomed1/import-1/rows.csv',
+  'execution/wsdoomed1/wf/ex/large-value-lv_1.json',
+  'table-snapshots/wsdoomed1/table-1/v1-hash.csv',
+  'copilot-doc-compiled/wsdoomed1/hash.pdf',
+  'workspace-logos/wsdoomed1/logo.png',
+]
+
+const NEIGHBOR_KEYS = [
+  'workspace/wsdoomed10/keep.txt',
+  'workspace/wsdoomed1-archive/keep.txt',
+  'execution/wsdoomed10/wf/ex/keep.json',
+  'project/wsdoomed1/keep.txt',
+  'kb/wsdoomed1/keep.txt',
+]
+
+describe('deleted workspace storage sweep', () => {
+  beforeEach(async () => {
+    resetDbChainMock()
+    root = await mkdtemp(join(tmpdir(), 'sim-account-sweep-'))
+    setUploadDirServer(root)
+    for (const key of [...DOOMED_KEYS, ...NEIGHBOR_KEYS]) await put(key)
+  })
+
+  afterEach(async () => {
+    await rm(root, { recursive: true, force: true })
+  })
+
+  it('deletes every family under the workspace prefix and nothing under a neighboring id', async () => {
+    await expect(sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())).resolves.toBe(
+      undefined
+    )
+
+    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(false)
+    for (const key of NEIGHBOR_KEYS) expect(await exists(key)).toBe(true)
+    expect(await exists('workspace/wsdoomed1')).toBe(false)
+  })
+
+  it('touches nothing while a workspace row with that id exists', async () => {
+    queueTableRows(schemaMock.workspace, [{ id: 'wsdoomed1' }])
+
+    await expect(sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())).resolves.toBe(
+      undefined
+    )
+
+    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(true)
+  })
+
+  it('keeps an object a surviving row still references', async () => {
+    queueTableRows(schemaMock.document, [{ key: 'workspace/wsdoomed1/a.txt' }])
+
+    await sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())
+
+    expect(await exists('workspace/wsdoomed1/a.txt')).toBe(true)
+    expect(await exists('workspace/wsdoomed1/nested/b.txt')).toBe(false)
+  })
+
+  it('stops deleting once the workspace reappears mid-sweep', async () => {
+    const keys = Array.from({ length: 510 }, (_, index) => `workspace/wsdoomed1/bulk/${index}.bin`)
+    for (const key of keys) await put(key)
+    queueTableRows(schemaMock.workspace, [])
+    queueTableRows(schemaMock.workspace, [])
+    queueTableRows(schemaMock.workspace, [{ id: 'wsdoomed1' }])
+
+    await expect(sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())).resolves.toBe(
+      undefined
+    )
+
+    const remaining = await Promise.all(keys.map(exists))
+    expect(remaining.filter(Boolean).length).toBeGreaterThan(0)
+    expect(await exists('workspace-logos/wsdoomed1/logo.png')).toBe(true)
+  })
+
+  it('keeps a public object a URL written mid-sweep names', async () => {
+    const logos = Array.from(
+      { length: 510 },
+      (_, index) => `workspace-logos/wsdoomed1/l/${String(index).padStart(3, '0')}.png`
+    )
+    for (const key of logos) await put(key)
+    queueTableRows(schemaMock.user, [])
+    queueTableRows(schemaMock.user, [
+      { id: 'user-9', value: '/api/files/serve/workspace-logos%2Fwsdoomed1%2Fl%2F505.png' },
+    ])
+
+    await sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())
+
+    expect(await exists('workspace-logos/wsdoomed1/l/505.png')).toBe(true)
+    expect(await exists('workspace-logos/wsdoomed1/l/504.png')).toBe(false)
+  })
+
+  it('walks past a full listing page within one run', async () => {
+    const keys = Array.from({ length: 510 }, (_, index) => `workspace/wsdoomed1/bulk/${index}.bin`)
+    for (const key of keys) await put(key)
+
+    await expect(sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())).resolves.toBe(
+      undefined
+    )
+
+    for (const key of keys) expect(await exists(key)).toBe(false)
+  })
+
+  it('resumes from its checkpointed family', async () => {
+    await sweep({ version: 1, workspaceId: 'wsdoomed1', family: 2 }, outboxContext())
+
+    expect(await exists('workspace/wsdoomed1/a.txt')).toBe(true)
+    expect(await exists('table-import/wsdoomed1/import-1/rows.csv')).toBe(true)
+    expect(await exists('execution/wsdoomed1/wf/ex/large-value-lv_1.json')).toBe(false)
+  })
+
+  it.each([
+    {
+      name: 'an organization whitelabel logo serve URL',
+      table: 'organization',
+      row: {
+        logo: null,
+        whitelabel: {
+          logoUrl: 'https://sim.test/api/files/serve/s3/workspace-logos%2Fwsdoomed1%2Flogo.png',
+        },
+        metadata: null,
+      },
+    },
+    {
+      name: 'a custom block icon stored as a direct bucket URL',
+      table: 'customBlock',
+      row: { value: 'https://bucket.s3.amazonaws.com/workspace-logos/wsdoomed1/logo.png' },
+    },
+    {
+      name: 'a deployed chat avatar',
+      table: 'chat',
+      row: {
+        value: { imageUrl: '/api/files/serve/workspace-logos/wsdoomed1/logo.png?context=chat' },
+      },
+    },
+    {
+      name: 'another account image',
+      table: 'user',
+      row: { value: '/api/files/serve/workspace-logos/wsdoomed1/logo.png' },
+    },
+  ] as const)('keeps an object named by $name', async ({ table, row }) => {
+    queueTableRows(schemaMock[table], [row])
+
+    await sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())
+
+    expect(await exists('workspace-logos/wsdoomed1/logo.png')).toBe(true)
+    expect(await exists('workspace/wsdoomed1/a.txt')).toBe(false)
+  })
+
+  it('keeps the local metadata sidecar of a kept object', async () => {
+    await put('workspace/wsdoomed1/a.txt.upload-metadata.json', '{}')
+    await put('workspace/wsdoomed1/nested/b.txt.upload-metadata.json', '{}')
+    queueTableRows(schemaMock.document, [{ key: 'workspace/wsdoomed1/a.txt' }])
+
+    await sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())
+
+    expect(await exists('workspace/wsdoomed1/a.txt')).toBe(true)
+    expect(await exists('workspace/wsdoomed1/a.txt.upload-metadata.json')).toBe(true)
+    expect(await exists('workspace/wsdoomed1/nested/b.txt.upload-metadata.json')).toBe(false)
+  })
+
+  it('pages past a full page of kept objects to the ones behind it', async () => {
+    const keptKeys = Array.from(
+      { length: 520 },
+      (_, index) => `workspace/wsdoomed1/a-kept/${String(index).padStart(3, '0')}.bin`
+    )
+    const unkeptKeys = Array.from(
+      { length: 10 },
+      (_, index) => `workspace/wsdoomed1/z-unkept/${index}.bin`
+    )
+    for (const key of [...keptKeys, ...unkeptKeys]) await put(key)
+    queueTableRows(
+      schemaMock.document,
+      keptKeys.map((key) => ({ key }))
+    )
+    queueTableRows(
+      schemaMock.document,
+      keptKeys.map((key) => ({ key }))
+    )
+
+    await expect(sweep({ version: 1, workspaceId: 'wsdoomed1' }, outboxContext())).resolves.toBe(
+      undefined
+    )
+
+    for (const key of unkeptKeys) expect(await exists(key)).toBe(false)
+    for (const key of keptKeys) expect(await exists(key)).toBe(true)
+  })
+
+  it.each([
+    { version: 1, workspaceId: '' },
+    { version: 1, workspaceId: '../wsdoomed1' },
+    { version: 1, workspaceId: 'wsdoomed1/' },
+    { version: 1, workspaceId: '*' },
+    { workspaceId: 'wsdoomed1' },
+  ])('rejects an unbounded or malformed payload %o', async (payload) => {
+    await expect(sweep(payload, outboxContext())).rejects.toThrow()
+    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(true)
+  })
+})
+
+describe('account storage object cleanup', () => {
+  beforeEach(async () => {
+    resetDbChainMock()
+    root = await mkdtemp(join(tmpdir(), 'sim-account-objects-'))
+    setUploadDirServer(root)
+  })
+
+  afterEach(async () => {
+    await rm(root, { recursive: true, force: true })
+  })
+
+  const payload = (objects: { key: string; context: string }[]) => ({
+    version: 1,
+    userId: 'user-1',
+    objects,
+  })
+
+  it('deletes unreferenced objects and keeps referenced ones', async () => {
+    await put('copilot/abc/private.txt')
+    await put('workspace/wsliving2/handed-off.txt')
+    queueTableRows(schemaMock.workspaceFiles, [{ key: 'workspace/wsliving2/handed-off.txt' }])
+
+    await cleanup(
+      payload([
+        { key: 'copilot/abc/private.txt', context: 'copilot' },
+        { key: 'workspace/wsliving2/handed-off.txt', context: 'workspace' },
+        { key: 'copilot/abc/already-gone.txt', context: 'copilot' },
+      ]),
+      outboxContext()
+    )
+
+    expect(await exists('copilot/abc/private.txt')).toBe(false)
+    expect(await readFile(join(root, 'workspace/wsliving2/handed-off.txt'), 'utf8')).toBe('x')
+  })
+
+  it('touches nothing while the account exists', async () => {
+    await put('copilot/abc/private.txt')
+    queueTableRows(schemaMock.user, [{ id: 'user-1' }])
+
+    await cleanup(
+      payload([{ key: 'copilot/abc/private.txt', context: 'copilot' }]),
+      outboxContext()
+    )
+
+    expect(await exists('copilot/abc/private.txt')).toBe(true)
+  })
+
+  it('fails so the outbox retries when a deletion fails, then completes', async () => {
+    await mkdir(join(root, 'copilot/abc/blocked.txt/child'), { recursive: true })
+    const objects = [{ key: 'copilot/abc/blocked.txt', context: 'copilot' }]
+
+    await expect(cleanup(payload(objects), outboxContext())).rejects.toThrow()
+
+    await rm(join(root, 'copilot/abc/blocked.txt'), { recursive: true })
+    await put('copilot/abc/blocked.txt')
+    await cleanup(payload(objects), outboxContext())
+    expect(await exists('copilot/abc/blocked.txt')).toBe(false)
+  })
+
+  it('keeps a key under a workspace that still exists', async () => {
+    await put('workspace/wsliving2/private.txt')
+    await put('copilot/abc/private.txt')
+    queueTableRows(schemaMock.workspace, [{ id: 'wsliving2' }])
+
+    await cleanup(
+      payload([
+        { key: 'workspace/wsliving2/private.txt', context: 'mothership' },
+        { key: 'copilot/abc/private.txt', context: 'copilot' },
+      ]),
+      outboxContext()
+    )
+
+    expect(await exists('workspace/wsliving2/private.txt')).toBe(true)
+    expect(await exists('copilot/abc/private.txt')).toBe(false)
+  })
+
+  it('keeps a publicly served key a stored asset URL names', async () => {
+    await put('workspace-logos/wsforeign9/icon.png')
+    queueTableRows(schemaMock.oauthClient, [
+      { value: '/api/files/serve/workspace-logos%2Fwsforeign9%2Ficon.png' },
+    ])
+
+    await cleanup(
+      payload([{ key: 'workspace-logos/wsforeign9/icon.png', context: 'workspace-logos' }]),
+      outboxContext()
+    )
+
+    expect(await exists('workspace-logos/wsforeign9/icon.png')).toBe(true)
+  })
+
+  it('keeps a public key without an owner id to narrow on, reading no URL column', async () => {
+    await put('og-images/logo.png')
+
+    await cleanup(payload([{ key: 'og-images/logo.png', context: 'workspace' }]), outboxContext())
+
+    expect(await exists('og-images/logo.png')).toBe(true)
+    for (const table of ['organization', 'customBlock', 'chat', 'oauthClient'] as const)
+      expect(dbChainMockFns.from).not.toHaveBeenCalledWith(schemaMock[table])
+  })
+
+  it("keeps an avatar another account's image still names", async () => {
+    const key = 'profile-pictures/upload-0002-owner-avatar.png'
+    await put(key)
+    queueTableRows(schemaMock.user, [])
+    queueTableRows(schemaMock.user, [
+      { id: 'owner-1', value: `/api/files/serve/s3/${encodeURIComponent(key)}` },
+    ])
+
+    await cleanup(payload([{ key, context: 'profile-pictures' }]), outboxContext())
+
+    expect(await exists(key)).toBe(true)
+  })
+
+  it("deletes the account's own uploaded avatar", async () => {
+    await put('profile-pictures/upload-0001-avatar.png')
+
+    await cleanup(
+      payload([{ key: 'profile-pictures/upload-0001-avatar.png', context: 'profile-pictures' }]),
+      outboxContext()
+    )
+
+    expect(await exists('profile-pictures/upload-0001-avatar.png')).toBe(false)
+  })
+
+  it('reads no URL column for keys that are never served without a row', async () => {
+    await put('copilot/abc/icon.png')
+    queueTableRows(schemaMock.oauthClient, [{ value: '/api/files/serve/copilot%2Fabc%2Ficon.png' }])
+
+    await cleanup(payload([{ key: 'copilot/abc/icon.png', context: 'copilot' }]), outboxContext())
+
+    expect(await exists('copilot/abc/icon.png')).toBe(false)
+    for (const table of ['organization', 'customBlock', 'chat', 'oauthClient'] as const)
+      expect(dbChainMockFns.from).not.toHaveBeenCalledWith(schemaMock[table])
+  })
+
+  it('still purges when a stored URL has malformed percent-encoding', async () => {
+    await put('workspace-logos/wsforeign9/old.png')
+    queueTableRows(schemaMock.user, [])
+    queueTableRows(schemaMock.user, [{ value: '/api/files/serve/%E0%A4%A' }])
+
+    await cleanup(
+      payload([{ key: 'workspace-logos/wsforeign9/old.png', context: 'workspace-logos' }]),
+      outboxContext()
+    )
+
+    expect(await exists('workspace-logos/wsforeign9/old.png')).toBe(false)
+  })
+
+  it('keeps a legacy key holding a literal percent sign that a malformed URL names', async () => {
+    const key = 'workspace-logos/wsforeign9/brand logo%.png'
+    const storedAs = 'workspace-logos/wsforeign9/brand_logo_.png'
+    await put(storedAs)
+    queueTableRows(schemaMock.user, [])
+    queueTableRows(schemaMock.user, [
+      { value: '/api/files/serve/workspace-logos/wsforeign9/brand%20logo%.png' },
+    ])
+
+    await cleanup(payload([{ key, context: 'workspace-logos' }]), outboxContext())
+
+    expect(await exists(storedAs)).toBe(true)
+  })
+
+  it.each([
+    { key: 'project/p1/file.txt', context: 'workspace' },
+    { key: 'assistant/org/user/file.png', context: 'mothership' },
+    { key: '../outside.txt', context: 'workspace' },
+    { key: 'workspace/wsliving2/file.txt', context: 'og-images' },
+    { key: 'profile-pictures/user-9/logo.png', context: 'workspace' },
+    { key: 'organization-logos/org-1/logo.png', context: 'workspace' },
+  ])('refuses a batch naming %o', async (object) => {
+    await expect(cleanup(payload([object]), outboxContext())).rejects.toThrow(
+      'Invalid account storage cleanup object'
+    )
+  })
+})
+
+describe('storage reference guard', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+  })
+
+  it('keeps a key under a workspace that still exists for every caller', async () => {
+    queueTableRows(schemaMock.workspace, [{ id: 'wsliving2' }])
+
+    const kept = await createStorageReferenceGuard()([
+      'execution/wsliving2/wf/ex/file.json',
+      'execution/wsdoomed1/wf/ex/file.json',
+    ])
+
+    expect([...kept]).toEqual(['execution/wsliving2/wf/ex/file.json'])
+  })
+})
+
+describe('queueing account storage cleanup', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+  })
+
+  async function queueCleanup(): Promise<void> {
+    const objects = await collectAccountStorageObjects({ userId: 'user-1', workspaceIds: [] })
+    await enqueueAccountStorageCleanupInTx(dbChainMock.db as unknown as DbTransaction, {
+      userId: 'user-1',
+      workspaceIds: [],
+      objects,
+    })
+  }
+
+  function queuedObjectKeys(): string[] {
+    return dbChainMockFns.values.mock.calls.flatMap(([rows]) =>
+      (rows as { payload: { objects?: { key: string }[] } }[]).flatMap(
+        (row) => row.payload.objects?.map((object) => object.key) ?? []
+      )
+    )
+  }
+
+  it('queues every version of a private file across version pages', async () => {
+    queueTableRows(schemaMock.workspaceFiles, [
+      { id: 'file-1', key: 'copilot/abc/file.txt', context: 'copilot' },
+    ])
+    const versionPage = (start: number, count: number) =>
+      Array.from({ length: count }, (_, index) => ({
+        id: `v-${String(start + index).padStart(5, '0')}`,
+        key: `copilot/abc/v${start + index}.txt`,
+        fileId: 'file-1',
+      }))
+    queueTableRows(schemaMock.workspaceFileVersion, versionPage(0, 1000))
+    queueTableRows(schemaMock.workspaceFileVersion, versionPage(1000, 5))
+
+    await queueCleanup()
+
+    const keys = queuedObjectKeys()
+    expect(keys).toHaveLength(1006)
+    expect(keys).toContain('copilot/abc/v1004.txt')
+  })
+
+  it('queues a file committed between the early read and the deletion locks, once', async () => {
+    const early = { id: 'file-1', key: 'copilot/abc/early.txt', context: 'copilot' }
+    queueTableRows(schemaMock.workspaceFiles, [early])
+    const objects = await collectAccountStorageObjects({ userId: 'user-1', workspaceIds: [] })
+    queueTableRows(schemaMock.workspaceFiles, [
+      early,
+      { id: 'file-2', key: 'copilot/abc/late.txt', context: 'copilot' },
+    ])
+
+    await enqueueAccountStorageCleanupInTx(dbChainMock.db as unknown as DbTransaction, {
+      userId: 'user-1',
+      workspaceIds: [],
+      objects,
+    })
+
+    expect(queuedObjectKeys()).toEqual(['copilot/abc/early.txt', 'copilot/abc/late.txt'])
+  })
+
+  it("queues the account's own uploaded avatar", async () => {
+    queueTableRows(schemaMock.user, [
+      { image: '/api/files/serve/s3/profile-pictures%2Fupload-0001-avatar.png' },
+    ])
+
+    await queueCleanup()
+
+    expect(queuedObjectKeys()).toEqual(['profile-pictures/upload-0001-avatar.png'])
+  })
+
+  it('never queues an avatar key that could escape its root', async () => {
+    queueTableRows(schemaMock.user, [
+      { image: '/api/files/serve/profile-pictures%2Fx%2F..%2F..%2Fworkspace%2Fother.png' },
+    ])
+
+    await queueCleanup()
+
+    expect(queuedObjectKeys()).toEqual([])
+  })
+
+  it('never queues a profile picture row', async () => {
+    queueTableRows(schemaMock.workspaceFiles, [
+      { id: 'file-1', key: 'profile-pictures/org-logo.png', context: 'profile-pictures' },
+      { id: 'file-2', key: 'copilot/abc/file.txt', context: 'copilot' },
+    ])
+
+    await queueCleanup()
+
+    expect(queuedObjectKeys()).toEqual(['copilot/abc/file.txt'])
+  })
+})
diff --git a/apps/sim/lib/users/storage-sweep.ts b/apps/sim/lib/users/storage-sweep.ts
new file mode 100644
index 00000000000..ef649fda616
--- /dev/null
+++ b/apps/sim/lib/users/storage-sweep.ts
@@ -0,0 +1,724 @@
+import { db } from '@sim/db'
+import {
+  copilotChats,
+  document,
+  knowledgeBase,
+  user,
+  workspaceFile,
+  workspaceFiles,
+  workspaceFileVersion,
+  workspace as workspaceTable,
+} from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { describeError } from '@sim/utils/errors'
+import { chunkArray } from '@sim/utils/helpers'
+import { isRecordLike } from '@sim/utils/object'
+import {
+  and,
+  eq,
+  gt,
+  gte,
+  inArray,
+  isNotNull,
+  isNull,
+  notInArray,
+  or,
+  type SQL,
+  type SQLWrapper,
+  sql,
+} from 'drizzle-orm'
+import {
+  continueOutboxHandler,
+  deferOutboxHandler,
+  enqueueOutboxEvents,
+  MAX_BULK_ENQUEUE_EVENTS,
+  type OutboxEventContext,
+  type OutboxHandler,
+  type OutboxHandlerRegistry,
+} from '@/lib/core/outbox/service'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import type { StorageContext } from '@/lib/uploads/config'
+import {
+  deleteStorageKeysUnderPrefix,
+  listStorageKeyPage,
+  outboxDeadlineNear,
+  storageCleanupAvailableAt,
+} from '@/lib/uploads/core/prefix-sweep'
+import { LOCAL_UPLOAD_METADATA_SUFFIX } from '@/lib/uploads/core/storage-key'
+import { deleteFile } from '@/lib/uploads/core/storage-service'
+import { WORKSPACE_SHARED_FILE_CONTEXTS } from '@/lib/uploads/shared/types'
+import { createStorageReferenceGuard } from '@/lib/users/storage-reference-guard'
+import {
+  ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT,
+  ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+} from '@/lib/users/storage-sweep-events'
+import { STORAGE_ID_PATTERN, WORKSPACE_STORAGE_FAMILIES } from '@/lib/users/workspace-storage'
+
+const logger = createLogger('AccountStorageSweep')
+
+const ROW_PAGE_SIZE = 1000
+const OBJECTS_PER_EVENT = 100
+
+/**
+ * Ceiling on exact keys one deletion records for durable cleanup. Past it the rest are logged
+ * and left; the workspace prefix sweeps are unaffected, since they list storage instead.
+ */
+const MAX_DURABLE_OBJECTS = 100_000
+
+const OBJECT_CONTEXTS: ReadonlySet<string> = new Set<StorageContext>([
+  'workspace',
+  'mothership',
+  'copilot',
+  'chat',
+  'knowledge-base',
+  'execution',
+  'table-import',
+  'workspace-logos',
+])
+
+/**
+ * Keys no row collected here may name: Project prefix cleanup, the Assistant upload-session
+ * sweep, organization logos, and profile pictures — which account deletion takes only from the
+ * account's own `user.image` (see {@link extractProfilePictureKey}), because whitelabel logos were
+ * once uploaded as `profile-pictures` rows under an admin's account.
+ */
+const FOREIGN_LIFECYCLE_ROOTS = [
+  'project/',
+  'assistant/',
+  'profile-pictures/',
+  'organization-logos/',
+] as const
+
+interface StorageObject {
+  key: string
+  context: StorageContext
+}
+
+interface WorkspaceSweepPayload {
+  workspaceId: string
+  family: number
+}
+
+interface ObjectCleanupPayload {
+  userId: string
+  objects: StorageObject[]
+  cursor: number
+}
+
+/** The trailing slash keeps `workspace/ws1/` from matching `workspace/ws10/...`. */
+function workspaceStoragePrefix(root: string, workspaceId: string): string {
+  if (!STORAGE_ID_PATTERN.test(workspaceId)) throw new Error('Invalid workspace storage identity')
+  return `${root}/${workspaceId}/`
+}
+
+function parseWorkspaceSweep(value: unknown): WorkspaceSweepPayload {
+  if (!isRecordLike(value) || value.version !== 1 || typeof value.workspaceId !== 'string')
+    throw new Error('Invalid workspace storage sweep payload')
+  workspaceStoragePrefix('workspace', value.workspaceId)
+  const family = value.family === undefined ? 0 : value.family
+  if (
+    typeof family !== 'number' ||
+    !Number.isSafeInteger(family) ||
+    family < 0 ||
+    family > WORKSPACE_STORAGE_FAMILIES.length
+  )
+    throw new Error('Invalid workspace storage sweep cursor')
+  return { workspaceId: value.workspaceId, family }
+}
+
+function isSafeKey(key: string): boolean {
+  return (
+    key.length > 0 && key.length <= 2048 && !key.startsWith('/') && !key.split('/').includes('..')
+  )
+}
+
+function isDisposableKey(key: string): boolean {
+  return isSafeKey(key) && !FOREIGN_LIFECYCLE_ROOTS.some((root) => key.startsWith(root))
+}
+
+function isAccountProfilePicture(key: string, context: string): boolean {
+  return context === 'profile-pictures' && key.startsWith('profile-pictures/')
+}
+
+/**
+ * The account's avatar, as a storage key. `user.image` holds either a `/api/files/serve/...` path
+ * (an upload this app owns) or an absolute URL from an OAuth provider, which must not be deleted;
+ * only the former yields a key, and only under `profile-pictures/`. The image is personal data, so
+ * an erasure that leaves it in the bucket is not an erasure. `user.image` is user-settable, so the
+ * key can name someone else's avatar; the cleanup's reference guard keeps any key another
+ * account's `user.image` or a stored asset URL still names.
+ */
+export function extractProfilePictureKey(image: string | null): string | null {
+  if (!image) return null
+  try {
+    const parsed = new URL(image, 'http://placeholder')
+    if (parsed.origin !== 'http://placeholder') return null
+    const segments = parsed.pathname.split('/')
+    if (segments[1] !== 'api' || segments[2] !== 'files' || segments[3] !== 'serve') return null
+    let keySegments = segments.slice(4)
+    if (['s3', 'blob', 'gcs'].includes(keySegments[0])) keySegments = keySegments.slice(1)
+    const key = decodeURIComponent(keySegments.join('/'))
+    return key.startsWith('profile-pictures/') ? key : null
+  } catch {
+    return null
+  }
+}
+
+function parseObjectCleanup(value: unknown): ObjectCleanupPayload {
+  if (
+    !isRecordLike(value) ||
+    value.version !== 1 ||
+    typeof value.userId !== 'string' ||
+    !STORAGE_ID_PATTERN.test(value.userId) ||
+    !Array.isArray(value.objects) ||
+    value.objects.length > OBJECTS_PER_EVENT
+  )
+    throw new Error('Invalid account storage cleanup payload')
+  const objects = value.objects.map((object): StorageObject => {
+    if (
+      !isRecordLike(object) ||
+      typeof object.key !== 'string' ||
+      typeof object.context !== 'string' ||
+      !(
+        (isDisposableKey(object.key) && OBJECT_CONTEXTS.has(object.context)) ||
+        (isSafeKey(object.key) && isAccountProfilePicture(object.key, object.context))
+      )
+    )
+      throw new Error('Invalid account storage cleanup object')
+    return { key: object.key, context: object.context as StorageContext }
+  })
+  const cursor = value.cursor === undefined ? 0 : value.cursor
+  if (
+    typeof cursor !== 'number' ||
+    !Number.isSafeInteger(cursor) ||
+    cursor < 0 ||
+    cursor > objects.length
+  )
+    throw new Error('Invalid account storage cleanup cursor')
+  return { userId: value.userId, objects, cursor }
+}
+
+/** A local sidecar belongs to its object: it is kept exactly when the object is. */
+function sidecarOwner(key: string): string | null {
+  return key.endsWith(LOCAL_UPLOAD_METADATA_SUFFIX)
+    ? key.slice(0, -LOCAL_UPLOAD_METADATA_SUFFIX.length)
+    : null
+}
+
+/**
+ * Walks one prefix to its end, deleting every listed object the reference guard does not keep,
+ * following the listing's page token within a run so kept objects never hold back later ones. A
+ * page that neither advances the token nor removes anything is `stalled` rather than re-listed
+ * forever.
+ */
+async function sweepWorkspacePrefix(
+  prefix: string,
+  storageContext: StorageContext,
+  context: OutboxEventContext,
+  isReferenced: (keys: readonly string[]) => Promise<Set<string>>,
+  workspaceExists: () => Promise<boolean>
+): Promise<{ deleted: number; status: 'done' | 'partial' | 'stalled' | 'refused' }> {
+  let deleted = 0
+  let pageToken: string | undefined
+  for (;;) {
+    if (outboxDeadlineNear(context)) return { deleted, status: 'partial' }
+    const page = await listStorageKeyPage(prefix, storageContext, context.signal, pageToken)
+    if (page.keys.some((key) => !key.startsWith(prefix)))
+      throw new Error('Storage provider returned a key outside the workspace prefix')
+    const owners = page.keys.flatMap((key) => sidecarOwner(key) ?? [])
+    const referenced = await isReferenced([...page.keys, ...owners])
+    const kept = (key: string) => referenced.has(key) || referenced.has(sidecarOwner(key) ?? key)
+    const keptCount = page.keys.filter(kept).length
+    if (keptCount > 0)
+      logger.warn('Keeping objects a surviving reference still names', { prefix, kept: keptCount })
+    if (await workspaceExists()) return { deleted, status: 'refused' }
+    const result = await deleteStorageKeysUnderPrefix(
+      prefix,
+      storageContext,
+      page.keys.filter((key) => !kept(key)),
+      context
+    )
+    deleted += result.deleted
+    if (!result.complete) return { deleted, status: 'partial' }
+    if (page.done) return { deleted, status: 'done' }
+    const advanced = page.nextPageToken !== undefined && page.nextPageToken !== pageToken
+    if (advanced) {
+      pageToken = page.nextPageToken
+    } else if (result.deleted === 0) {
+      logger.error('Workspace storage sweep made no progress on a full listing page', { prefix })
+      return { deleted, status: 'stalled' }
+    }
+  }
+}
+
+/**
+ * Deletes a deleted workspace's objects by prefix in each bucket. A workspace row with the same
+ * id means the database no longer agrees the workspace is gone (a restore, a point-in-time
+ * recovery), so the sweep stops without touching storage; it looks again before every page it
+ * deletes from.
+ */
+const sweepWorkspaceStorage: OutboxHandler<unknown> = async (raw, context) => {
+  const payload = parseWorkspaceSweep(raw)
+  const workspaceExists = async () => {
+    const [existing] = await db
+      .select({ id: workspaceTable.id })
+      .from(workspaceTable)
+      .where(eq(workspaceTable.id, payload.workspaceId))
+      .limit(1)
+    if (existing)
+      logger.error('Refusing to sweep storage for a workspace that exists', {
+        workspaceId: payload.workspaceId,
+      })
+    return Boolean(existing)
+  }
+  if (await workspaceExists()) return
+  const isReferenced = createStorageReferenceGuard()
+  let progressed = false
+  for (let family = payload.family; family < WORKSPACE_STORAGE_FAMILIES.length; family++) {
+    const { root, context: storageContext } = WORKSPACE_STORAGE_FAMILIES[family]
+    const prefix = workspaceStoragePrefix(root, payload.workspaceId)
+    const result = await sweepWorkspacePrefix(
+      prefix,
+      storageContext,
+      context,
+      isReferenced,
+      workspaceExists
+    )
+    if (result.status === 'refused') return
+    if (result.deleted > 0) progressed = true
+    if (result.status === 'stalled')
+      return deferOutboxHandler('Workspace storage sweep made no progress on a full listing page')
+    if (result.status === 'partial')
+      return progressed
+        ? continueOutboxHandler('Continuing workspace storage sweep')
+        : deferOutboxHandler('Workspace storage sweep made no progress before its deadline')
+    await context.checkpointPayload({ family: family + 1 })
+    progressed = true
+  }
+}
+
+/**
+ * Deletes exact keys recorded by an account deletion. The account reappearing refuses the batch
+ * as a whole; a key the reference guard keeps, or one under a workspace that still exists, stays.
+ */
+const cleanupAccountObjects: OutboxHandler<unknown> = async (raw, context) => {
+  const payload = parseObjectCleanup(raw)
+  const [existing] = await db
+    .select({ id: user.id })
+    .from(user)
+    .where(eq(user.id, payload.userId))
+    .limit(1)
+  if (existing) {
+    logger.error('Refusing to delete storage for an account that exists', {
+      userId: payload.userId,
+    })
+    return
+  }
+  const pending = payload.objects.slice(payload.cursor).map((object) => object.key)
+  const referenced = await createStorageReferenceGuard()(pending)
+  if (referenced.size > 0)
+    logger.warn('Keeping objects a surviving reference still names', {
+      userId: payload.userId,
+      kept: referenced.size,
+    })
+  for (let index = payload.cursor; index < payload.objects.length; index++) {
+    context.signal.throwIfAborted()
+    if (index > payload.cursor && outboxDeadlineNear(context)) {
+      await context.checkpointPayload({ cursor: index })
+      return continueOutboxHandler('Continuing account storage cleanup')
+    }
+    const { key, context: storageContext } = payload.objects[index]
+    if (referenced.has(key)) continue
+    try {
+      await deleteFile({ key, context: storageContext, signal: context.signal })
+    } catch (error) {
+      if (describeError(error).code !== 'ENOENT') throw error
+    }
+  }
+}
+
+export const accountStorageSweepOutboxHandlers = {
+  [ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT]: sweepWorkspaceStorage,
+  [ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT]: cleanupAccountObjects,
+} satisfies OutboxHandlerRegistry
+
+/** True when `key` sits under one of `workspaceId`'s swept prefixes. */
+function coveredByWorkspaceSweep(key: SQLWrapper, workspaceId: SQLWrapper): SQL {
+  return sql`(${sql.join(
+    WORKSPACE_STORAGE_FAMILIES.map(
+      ({ root }) => sql`starts_with(${key}, ${`${root}/`} || ${workspaceId} || '/')`
+    ),
+    sql` OR `
+  )})`
+}
+
+interface ObjectRow {
+  id: string
+  key: string | null
+  context: string | null
+}
+
+/**
+ * How far before its read the pre-transaction collection's in-transaction delta looks back, so
+ * clock skew between this host and the database, and an insert whose transaction began earlier,
+ * cannot slip between the two reads. Re-reading a row only collects its key once.
+ */
+const DELTA_LOOKBACK_MS = 5 * 60_000
+
+/**
+ * Exact keys an account deletion hands to durable cleanup, bounded by {@link MAX_DURABLE_OBJECTS},
+ * and the moment from which the in-transaction delta must look for rows the read missed.
+ */
+interface AccountStorageObjects {
+  readonly objects: readonly StorageObject[]
+  readonly skippedContexts: number
+  readonly capped: boolean
+  readonly since: Date
+}
+
+class ObjectCollector implements AccountStorageObjects {
+  readonly objects: StorageObject[] = []
+  private readonly seen = new Set<string>()
+  skippedContexts = 0
+  capped = false
+
+  constructor(readonly since: Date) {}
+
+  static from(collected: AccountStorageObjects): ObjectCollector {
+    const collector = new ObjectCollector(collected.since)
+    collector.skippedContexts = collected.skippedContexts
+    collector.capped = collected.capped
+    for (const object of collected.objects) collector.push(object)
+    return collector
+  }
+
+  addProfilePicture(key: string): void {
+    if (!isSafeKey(key) || !isAccountProfilePicture(key, 'profile-pictures') || this.seen.has(key))
+      return
+    if (this.objects.length >= MAX_DURABLE_OBJECTS) {
+      this.capped = true
+      return
+    }
+    this.push({ key, context: 'profile-pictures' })
+  }
+
+  add(rows: readonly ObjectRow[]): void {
+    for (const row of rows) {
+      if (!row.key || !isDisposableKey(row.key)) continue
+      if (!row.context || !OBJECT_CONTEXTS.has(row.context)) {
+        this.skippedContexts++
+        continue
+      }
+      if (this.seen.has(row.key)) continue
+      if (this.objects.length >= MAX_DURABLE_OBJECTS) {
+        this.capped = true
+        return
+      }
+      this.push({ key: row.key, context: row.context as StorageContext })
+    }
+  }
+
+  private push(object: StorageObject): void {
+    this.seen.add(object.key)
+    this.objects.push(object)
+  }
+}
+
+async function collectPages(
+  collector: ObjectCollector,
+  page: (afterId: string) => Promise<ObjectRow[]>,
+  afterPage?: (rows: ObjectRow[]) => Promise<void>
+): Promise<void> {
+  let afterId = ''
+  while (!collector.capped) {
+    const rows = await page(afterId)
+    if (rows.length === 0) return
+    collector.add(rows)
+    if (afterPage) await afterPage(rows)
+    if (rows.length < ROW_PAGE_SIZE) return
+    afterId = rows[rows.length - 1].id
+  }
+}
+
+async function collectVersionsOf(
+  executor: DbOrTx,
+  collector: ObjectCollector,
+  files: readonly ObjectRow[]
+): Promise<void> {
+  if (files.length === 0) return
+  const contexts = new Map(files.map((file) => [file.id, file.context]))
+  const fileIds = files.map((file) => file.id)
+  await collectPages(collector, (afterId) =>
+    executor
+      .select({
+        id: workspaceFileVersion.id,
+        key: workspaceFileVersion.key,
+        fileId: workspaceFileVersion.fileId,
+      })
+      .from(workspaceFileVersion)
+      .where(
+        and(inArray(workspaceFileVersion.fileId, fileIds), gt(workspaceFileVersion.id, afterId))
+      )
+      .orderBy(workspaceFileVersion.id)
+      .limit(ROW_PAGE_SIZE)
+      .then((versions) =>
+        versions.map((version) => ({
+          id: version.id,
+          key: version.key,
+          context: contexts.get(version.fileId) ?? null,
+        }))
+      )
+  )
+}
+
+interface CollectionScope {
+  userId: string
+  workspaceIds: string[]
+  /** Set for the in-transaction delta: only the account's rows written since then. */
+  since?: Date
+}
+
+/**
+ * The exact-key reads behind {@link collectAccountStorageObjects}. With `since`, each read is
+ * narrowed to the account's own rows written from then on (`workspace_files.user_id`,
+ * `document.uploaded_by`, both indexed), and the doomed workspaces' version and legacy-file
+ * tables are skipped: a content write in a workspace stores its bytes under `workspace/<id>/`,
+ * which the prefix sweep lists.
+ */
+async function collectInto(
+  executor: DbOrTx,
+  collector: ObjectCollector,
+  { userId, workspaceIds, since }: CollectionScope
+): Promise<void> {
+  const [account] = await executor
+    .select({ image: user.image })
+    .from(user)
+    .where(eq(user.id, userId))
+    .limit(1)
+  const profilePicture = extractProfilePictureKey(account?.image ?? null)
+  if (profilePicture) collector.addProfilePicture(profilePicture)
+
+  const filesSince = since
+    ? and(eq(workspaceFiles.userId, userId), gte(workspaceFiles.uploadedAt, since))
+    : undefined
+  if (workspaceIds.length > 0) {
+    await collectPages(collector, (afterId) =>
+      executor
+        .select({ id: workspaceFiles.id, key: workspaceFiles.key, context: workspaceFiles.context })
+        .from(workspaceFiles)
+        .where(
+          and(
+            inArray(workspaceFiles.workspaceId, workspaceIds),
+            isNull(workspaceFiles.projectId),
+            filesSince,
+            sql`NOT ${coveredByWorkspaceSweep(workspaceFiles.key, workspaceFiles.workspaceId)}`,
+            gt(workspaceFiles.id, afterId)
+          )
+        )
+        .orderBy(workspaceFiles.id)
+        .limit(ROW_PAGE_SIZE)
+    )
+    if (!since) {
+      await collectPages(collector, (afterId) =>
+        executor
+          .select({
+            id: workspaceFileVersion.id,
+            key: workspaceFileVersion.key,
+            context: workspaceFiles.context,
+          })
+          .from(workspaceFileVersion)
+          .innerJoin(workspaceFiles, eq(workspaceFiles.id, workspaceFileVersion.fileId))
+          .where(
+            and(
+              inArray(workspaceFileVersion.workspaceId, workspaceIds),
+              isNull(workspaceFiles.projectId),
+              sql`NOT ${coveredByWorkspaceSweep(workspaceFileVersion.key, workspaceFileVersion.workspaceId)}`,
+              gt(workspaceFileVersion.id, afterId)
+            )
+          )
+          .orderBy(workspaceFileVersion.id)
+          .limit(ROW_PAGE_SIZE)
+      )
+      await collectPages(collector, (afterId) =>
+        executor
+          .select({
+            id: workspaceFile.id,
+            key: workspaceFile.key,
+            context: sql<string>`'workspace'`,
+          })
+          .from(workspaceFile)
+          .where(
+            and(
+              inArray(workspaceFile.workspaceId, workspaceIds),
+              sql`NOT ${coveredByWorkspaceSweep(workspaceFile.key, workspaceFile.workspaceId)}`,
+              gt(workspaceFile.id, afterId)
+            )
+          )
+          .orderBy(workspaceFile.id)
+          .limit(ROW_PAGE_SIZE)
+      )
+    }
+    await collectPages(collector, (afterId) =>
+      executor
+        .select({
+          id: document.id,
+          key: document.storageKey,
+          context: sql<string>`'knowledge-base'`,
+        })
+        .from(document)
+        .innerJoin(knowledgeBase, eq(knowledgeBase.id, document.knowledgeBaseId))
+        .where(
+          and(
+            inArray(knowledgeBase.workspaceId, workspaceIds),
+            isNotNull(document.storageKey),
+            since
+              ? and(eq(document.uploadedBy, userId), gte(document.uploadedAt, since))
+              : undefined,
+            sql`NOT ${coveredByWorkspaceSweep(document.storageKey, knowledgeBase.workspaceId)}`,
+            gt(document.id, afterId)
+          )
+        )
+        .orderBy(document.id)
+        .limit(ROW_PAGE_SIZE)
+    )
+  }
+
+  const ownChats = executor
+    .select({ id: copilotChats.id })
+    .from(copilotChats)
+    .where(eq(copilotChats.userId, userId))
+  await collectPages(
+    collector,
+    (afterId) =>
+      executor
+        .select({ id: workspaceFiles.id, key: workspaceFiles.key, context: workspaceFiles.context })
+        .from(workspaceFiles)
+        .where(
+          and(
+            eq(workspaceFiles.userId, userId),
+            filesSince,
+            isNull(workspaceFiles.projectId),
+            isNull(workspaceFiles.organizationId),
+            workspaceIds.length > 0
+              ? or(
+                  isNull(workspaceFiles.workspaceId),
+                  notInArray(workspaceFiles.workspaceId, workspaceIds)
+                )
+              : undefined,
+            or(
+              and(
+                isNull(workspaceFiles.workspaceId),
+                eq(workspaceFiles.context, 'copilot'),
+                or(isNull(workspaceFiles.chatId), inArray(workspaceFiles.chatId, ownChats))
+              ),
+              and(
+                isNotNull(workspaceFiles.workspaceId),
+                or(
+                  inArray(workspaceFiles.chatId, ownChats),
+                  and(
+                    isNull(workspaceFiles.chatId),
+                    notInArray(workspaceFiles.context, [...WORKSPACE_SHARED_FILE_CONTEXTS])
+                  )
+                )
+              )
+            ),
+            gt(workspaceFiles.id, afterId)
+          )
+        )
+        .orderBy(workspaceFiles.id)
+        .limit(ROW_PAGE_SIZE),
+    (files) => collectVersionsOf(executor, collector, files)
+  )
+}
+
+function sweepableWorkspaceIds(workspaceIds: readonly string[]): string[] {
+  return workspaceIds.filter((id) => STORAGE_ID_PATTERN.test(id))
+}
+
+/**
+ * Reads the exact keys an account deletion hands to durable cleanup, before its transaction, so
+ * these unbounded keyset reads never run while the deletion holds workspace locks that
+ * teammates' writes in shared workspaces wait on. Covers what cascades away outside the
+ * workspace prefix sweeps: objects in deleted workspaces outside those prefixes (knowledge-base
+ * documents, Copilot uploads), and the account's private files elsewhere — chat attachments of
+ * its own chats, private contexts, and Copilot files with no workspace. Files handed to a
+ * successor, Project and organization files, attachments of somebody else's chat, profile
+ * pictures, and workspace-less rows of any other context (a legacy shared row can lack its
+ * workspace) are never collected.
+ *
+ * A row committed after this read is picked up by the narrow delta
+ * {@link enqueueAccountStorageCleanupInTx} reads under the deletion's locks.
+ */
+export async function collectAccountStorageObjects(params: {
+  userId: string
+  workspaceIds: readonly string[]
+}): Promise<AccountStorageObjects> {
+  const collector = new ObjectCollector(new Date(Date.now() - DELTA_LOOKBACK_MS))
+  await collectInto(db, collector, {
+    userId: params.userId,
+    workspaceIds: sweepableWorkspaceIds(params.workspaceIds),
+  })
+  return collector
+}
+
+/**
+ * Queues the durable storage cleanup for an account deletion inside its transaction, so the work
+ * commits or rolls back with the deletion: one prefix sweep per deleted workspace, covering every
+ * object under its keyed families whether or not a row ever named it, and delayed events for the
+ * exact keys {@link collectAccountStorageObjects} read beforehand plus a delta of the account's
+ * own rows written since. The delta runs after the account row is locked, so no further row
+ * naming the account can commit; it reads through the `user_id` and `uploaded_by` indexes and
+ * takes no locks of its own.
+ *
+ * Both wait out every upload URL signed before the deletion and re-check at run time that the
+ * owner is still gone and that {@link createStorageReferenceGuard} does not keep the key; the
+ * exact-key cleanup also keeps any key under a workspace that still exists.
+ */
+export async function enqueueAccountStorageCleanupInTx(
+  tx: DbTransaction,
+  params: { userId: string; workspaceIds: readonly string[]; objects: AccountStorageObjects }
+): Promise<void> {
+  const { userId } = params
+  const availableAt = storageCleanupAvailableAt()
+  const workspaceIds = sweepableWorkspaceIds(params.workspaceIds)
+  if (workspaceIds.length !== params.workspaceIds.length) {
+    logger.error('Skipping storage sweeps for workspaces with unsafe ids', {
+      userId,
+      skipped: params.workspaceIds.length - workspaceIds.length,
+    })
+  }
+  for (const batch of chunkArray(workspaceIds, MAX_BULK_ENQUEUE_EVENTS)) {
+    await enqueueOutboxEvents(
+      tx,
+      ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+      batch.map((workspaceId) => ({ version: 1, workspaceId })),
+      { availableAt }
+    )
+  }
+
+  const collector = ObjectCollector.from(params.objects)
+  const collected = collector.objects.length
+  await collectInto(tx, collector, { userId, workspaceIds, since: collector.since })
+  if (collector.capped) {
+    logger.error('Account storage cleanup hit its key ceiling; the remainder is left', {
+      userId,
+      cap: MAX_DURABLE_OBJECTS,
+    })
+  }
+  const payloads = chunkArray(collector.objects, OBJECTS_PER_EVENT).map((chunk) => ({
+    version: 1,
+    userId,
+    objects: chunk,
+  }))
+  for (const batch of chunkArray(payloads, MAX_BULK_ENQUEUE_EVENTS)) {
+    await enqueueOutboxEvents(tx, ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT, batch, { availableAt })
+  }
+  logger.info('Queued account storage cleanup', {
+    userId,
+    workspaceSweeps: workspaceIds.length,
+    recorded: collector.objects.length,
+    fromDelta: collector.objects.length - collected,
+    skippedContexts: collector.skippedContexts,
+    capped: collector.capped,
+  })
+}
diff --git a/apps/sim/lib/users/workspace-storage.ts b/apps/sim/lib/users/workspace-storage.ts
new file mode 100644
index 00000000000..0c4d54c2af2
--- /dev/null
+++ b/apps/sim/lib/users/workspace-storage.ts
@@ -0,0 +1,54 @@
+import { db } from '@sim/db'
+import { workspace as workspaceTable } from '@sim/db/schema'
+import { inArray } from 'drizzle-orm'
+import type { StorageContext } from '@/lib/uploads/config'
+
+/** An id safe to place in a storage key path or a `LIKE` pattern. */
+export const STORAGE_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/
+
+/**
+ * Every storage family whose keys embed their workspace as the second path segment, in the
+ * bucket that serves it. A deleted workspace's objects in these families are found by listing,
+ * so neither a missing row nor a row written after the deletion read its keys can strand one.
+ */
+export const WORKSPACE_STORAGE_FAMILIES = [
+  { root: 'workspace', context: 'workspace' },
+  { root: 'table-import', context: 'table-import' },
+  { root: 'execution', context: 'execution' },
+  { root: 'table-snapshots', context: 'execution' },
+  { root: 'copilot-doc-compiled', context: 'copilot' },
+  { root: 'workspace-logos', context: 'workspace-logos' },
+] as const satisfies readonly { root: string; context: StorageContext }[]
+
+const WORKSPACE_KEYED_ROOTS: ReadonlySet<string> = new Set(
+  WORKSPACE_STORAGE_FAMILIES.map(({ root }) => root)
+)
+
+/**
+ * Exact keys under a workspace-keyed root whose workspace still exists. Such an object stays
+ * readable to that workspace's members by key, so no account-deletion purge deletes it, even when
+ * the row that named it cascaded away with the account. A workspace segment that is not a valid id
+ * is kept too.
+ */
+export async function keysUnderLiveWorkspaces(keys: readonly string[]): Promise<Set<string>> {
+  const kept = new Set<string>()
+  const keysByWorkspace = new Map<string, string[]>()
+  for (const key of keys) {
+    const [root, workspaceId] = key.split('/')
+    if (!WORKSPACE_KEYED_ROOTS.has(root)) continue
+    if (!workspaceId || !STORAGE_ID_PATTERN.test(workspaceId)) {
+      kept.add(key)
+      continue
+    }
+    const workspaceKeys = keysByWorkspace.get(workspaceId)
+    if (workspaceKeys) workspaceKeys.push(key)
+    else keysByWorkspace.set(workspaceId, [key])
+  }
+  if (keysByWorkspace.size === 0) return kept
+  const live = await db
+    .select({ id: workspaceTable.id })
+    .from(workspaceTable)
+    .where(inArray(workspaceTable.id, [...keysByWorkspace.keys()]))
+  for (const { id } of live) for (const key of keysByWorkspace.get(id) ?? []) kept.add(key)
+  return kept
+}
diff --git a/apps/sim/lib/workspaces/resource-handoff.ts b/apps/sim/lib/workspaces/resource-handoff.ts
index ffbfcc1bac0..4b12ba56cc8 100644
--- a/apps/sim/lib/workspaces/resource-handoff.ts
+++ b/apps/sim/lib/workspaces/resource-handoff.ts
@@ -20,20 +20,12 @@ import {
   lockProjectsForResourceDepartureTx,
 } from '@/lib/projects/resource-handoff'
 import { handoffFileCreatorsInTx } from '@/lib/uploads/contexts/workspace/creator-handoff'
+import { WORKSPACE_SHARED_FILE_CONTEXTS } from '@/lib/uploads/shared/types'
 import {
   findWorkspaceBillingReplacement,
   reassignWorkflowOwnershipForWorkspaceMemberRemovalTx,
 } from '@/lib/workspaces/utils'
 
-/** Private chat attachments keep their user's lifecycle even when stored under a workspace key. */
-const SHARED_FILE_CONTEXTS = [
-  'workspace',
-  'knowledge-base',
-  'execution',
-  'chat',
-  'workspace-logos',
-] as const
-
 /** Finds surviving containers from ownership references, including departures predating handoff. */
 export async function listSharedResourceWorkspaceIdsForUser(executor: DbOrTx, userId: string) {
   const rows = await executor.execute<{ workspaceId: string }>(sql`
@@ -43,7 +35,7 @@ export async function listSharedResourceWorkspaceIdsForUser(executor: DbOrTx, us
       UNION ALL SELECT ${userTableDefinitions.workspaceId} FROM ${userTableDefinitions} WHERE ${userTableDefinitions.createdBy} = ${userId}
       UNION ALL SELECT ${knowledgeBase.workspaceId} FROM ${knowledgeBase} WHERE ${knowledgeBase.userId} = ${userId}
       UNION ALL SELECT ${workspaceFiles.workspaceId} FROM ${workspaceFiles}
-        WHERE ${workspaceFiles.userId} = ${userId} AND ${inArray(workspaceFiles.context, SHARED_FILE_CONTEXTS)} AND ${workspaceFiles.chatId} IS NULL
+        WHERE ${workspaceFiles.userId} = ${userId} AND ${inArray(workspaceFiles.context, WORKSPACE_SHARED_FILE_CONTEXTS)} AND ${workspaceFiles.chatId} IS NULL
       UNION ALL SELECT ${workspaceFile.workspaceId} FROM ${workspaceFile} WHERE ${workspaceFile.uploadedBy} = ${userId}
       UNION ALL SELECT ${workflowMcpServer.workspaceId} FROM ${workflowMcpServer}
         WHERE ${workflowMcpServer.createdBy} = ${userId} AND NOT ${workflowMcpServer.isPublic}
@@ -206,7 +198,7 @@ export async function reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx
       and(
         eq(workspaceFiles.workspaceId, row.id),
         eq(workspaceFiles.userId, departingUserId),
-        inArray(workspaceFiles.context, SHARED_FILE_CONTEXTS),
+        inArray(workspaceFiles.context, WORKSPACE_SHARED_FILE_CONTEXTS),
         isNull(workspaceFiles.chatId)
       ),
       userId
diff --git a/packages/testing/src/mocks/file-utils.mock.ts b/packages/testing/src/mocks/file-utils.mock.ts
index 2a7bd1d7a15..f1878846366 100644
--- a/packages/testing/src/mocks/file-utils.mock.ts
+++ b/packages/testing/src/mocks/file-utils.mock.ts
@@ -576,6 +576,10 @@ export const fileUtilsMockFns = {
   mockInferContextFromKey: vi.fn(inferContextFromKey),
   mockTryInferContextFromKey: vi.fn(tryInferContextFromKey),
   mockIsPublicStorageContext: vi.fn(isPublicStorageContext),
+  mockIsPublicStorageKey: vi.fn((key: string) => {
+    const context = tryInferContextFromKey(key)
+    return context !== null && isPublicStorageContext(context)
+  }),
   mockResolveTrustedFileContext: vi.fn((key: string, context?: string): string => {
     try {
       return inferContextFromKey(key)
@@ -735,6 +739,7 @@ export const fileUtilsMock = {
   inferContextFromKey: fileUtilsMockFns.mockInferContextFromKey,
   tryInferContextFromKey: fileUtilsMockFns.mockTryInferContextFromKey,
   isPublicStorageContext: fileUtilsMockFns.mockIsPublicStorageContext,
+  isPublicStorageKey: fileUtilsMockFns.mockIsPublicStorageKey,
   resolveTrustedFileContext: fileUtilsMockFns.mockResolveTrustedFileContext,
   parseInternalFileUrl: fileUtilsMockFns.mockParseInternalFileUrl,
   resolveHttpsUrlFromFileInput: fileUtilsMockFns.mockResolveHttpsUrlFromFileInput,