@@ -45,6 +45,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 {
@@ -239,6 +243,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 () => {
 
@@ -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(),
+    ])
+  })
+})
@@ -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(),
 
@@ -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)
 
@@ -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 })
+  })
+})