chore(db): drop the legacy folder tables and adopt the deferred folder_id FKs (#6051)

* chore(db): drop the legacy folder tables and adopt the deferred folder_id FKs

Contract migration closing out the generic-folders cutover: drops workflow_folder
and workspace_file_folders, and adopts the two folder_id foreign keys 0272
deliberately deferred.

Runs a final INSERT-ONLY reconcile first so no stranded legacy folder is lost by
the drop. Insert-only because 0272 deliberately renamed 47 workflow folders to
dedupe them, and an upsert would revert all 47. In production this is a no-op
(verified 0 stranded); it exists for deployments that never ran the post-drain
reconcile as an operational step, self-hosted upgrades above all.

Hand-written rather than drizzle-generated: the generated form drops the tables
BEFORE adding the FKs with no step to re-root unresolvable folder_ids, so one
dangling id fails ADD CONSTRAINT after the legacy data is already gone; it also
validates the FK inline, holding ACCESS EXCLUSIVE across a full scan of the
1.7M-row workspace_files, and uses DROP TABLE CASCADE. The generated snapshot is
kept so drizzle-kit generate reports no schema changes.

Every re-root path dedupes, because all three destinations are partial unique
indexes keyed on a coalesced nullable column: folder, workflow, and
workspace_files. Parents must also be reachable — an active row re-roots off a
soft-deleted parent, while a soft-deleted row may keep one.

Also hardens the retention sweep, which the new ON DELETE SET NULL exposes: a
surviving active child re-rooted by the FK can collide at the workspace root,
and chunkedBatchDelete turns that 23505 into a permanent per-chunk stall. The
folder cleanup target now renames children first, covering workflows, files and
subfolders, re-asserts eligibility so a restore mid-batch is not stripped, and
treats the deduplicated name as a hint since both allocators can return a
colliding one.

* fix(db): stop the re-root dedupe renaming personal workflows

workflow.workspace_id is nullable, and NULL is treated as EQUAL by PARTITION BY
but UNKNOWN by the = in base_taken. So for personal workflows rn incremented
across the group while no collision was ever detected: the dedupe could only
fire spuriously, renaming a user-visible workflow that needed no rename, since
the unique index treats NULL workspace_id rows as distinct anyway. The file
block already carried the equivalent guard.

Also gives step 1's inserts ON CONFLICT (id) DO NOTHING. It restates the
stranded guard, closing the window between that read's snapshot and the index
check — an operational re-run of 0274, or a live pod, committing into folder
mid-statement would otherwise raise 23505, and migrate.ts retries only 55P03.
0272 and 0274 were already written this way; step 1 was the exception.

Corrects three comments that claimed more than the code delivers: the header's
'no legacy folder is lost' (an id already present under another resource_type
cannot be rescued), the reachability note (a cycle among stranded rows survives
a per-row parent check), and a note made stale by the ON CONFLICT above.

apps/sim/background/cleanup-soft-deletes.test.ts

@@ -2,7 +2,13 @@
  * @vitest-environment node
  */
 
-import { dbChainMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing'
+import {
+  dbChainMock,
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  schemaMock,
+} from '@sim/testing'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
@@ -18,7 +24,13 @@ const {
   mockPrepareChatCleanup,
   mockResolveStorageBillingContext,
   mockSelectRowsByIdChunks,
+  mockDeduplicateWorkflowName,
+  mockAllocateUniqueWorkspaceFileName,
+  mockDeduplicateFolderName,
 } = vi.hoisted(() => ({
+  mockDeduplicateFolderName: vi.fn(async (_tx, _ws, _parent, name: string) => name),
+  mockDeduplicateWorkflowName: vi.fn(async (name: string) => name),
+  mockAllocateUniqueWorkspaceFileName: vi.fn(async (_ws: string, name: string) => name),
   mockBatchDeleteByWorkspaceAndTimestamp: vi.fn(async () => ({ deleted: 0, failed: 0 })),
   mockChunkedBatchDelete: vi.fn(async () => ({ deleted: 0, failed: 0 })),
   mockDecrementStorageUsageForBillingContextInTx: vi.fn(async () => undefined),
@@ -66,6 +78,16 @@ vi.mock('@/lib/uploads', () => ({
 
 vi.mock('@/lib/uploads/server/metadata', () => ({ deleteFileMetadata: mockDeleteFileMetadata }))
 
+vi.mock('@/lib/workflows/utils', () => ({
+  deduplicateWorkflowName: mockDeduplicateWorkflowName,
+}))
+
+vi.mock('@/lib/folders/naming', () => ({ deduplicateFolderName: mockDeduplicateFolderName }))
+
+vi.mock('@/lib/uploads/contexts/workspace/workspace-file-manager', () => ({
+  allocateUniqueWorkspaceFileName: mockAllocateUniqueWorkspaceFileName,
+}))
+
 import { runCleanupSoftDeletes } from '@/background/cleanup-soft-deletes'
 
 const basePayload = {
@@ -269,6 +291,7 @@ interface BatchDeleteOptions {
   tableName: string
   requireTimestampNotNull?: boolean
   additionalPredicate?: { type: string; column: unknown; values: unknown[] }
+  onBatch?: (rows: { id: string }[]) => Promise<void>
 }
 
 /**
@@ -331,4 +354,177 @@ describe('folder cleanup target', () => {
       .map(([options]) => options.tableName)
     expect(filtered).toEqual(['free/1/folder'])
   })
+
+  /**
+   * `folder_id` is `ON DELETE SET NULL`, so Postgres re-roots surviving children on its own —
+   * but `workflow` and `workspace_files` each carry a partial unique index keyed on
+   * `coalesce(folder_id, '')`, so an implicit SET NULL can land a child on a name the workspace
+   * root already holds. That aborts the whole DELETE with a 23505, which `chunkedBatchDelete`
+   * turns into `hasMore = false` — folder retention then stalls permanently for that chunk,
+   * re-failing on every later run. `onBatch` renames first so the SET NULL is a no-op.
+   */
+  describe('re-rooting active children before the delete', () => {
+    async function getFolderOnBatch() {
+      const target = await runAndFindFolderTarget()
+      expect(target?.onBatch).toBeTypeOf('function')
+      return target!.onBatch!
+    }
+
+    it('is the only cleanup target that re-roots children', async () => {
+      await runCleanupSoftDeletes(basePayload)
+      const calls = mockBatchDeleteByWorkspaceAndTimestamp.mock.calls as unknown as Array<
+        [BatchDeleteOptions]
+      >
+
+      const withOnBatch = calls
+        .filter(([options]) => options.onBatch !== undefined)
+        .map(([options]) => options.tableName)
+      expect(withOnBatch).toEqual(['free/1/folder'])
+    })
+
+    it('re-roots an active workflow under a deduplicated name', async () => {
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, [{ id: 'folder-1' }])
+      queueTableRows(schemaMock.workflow, [{ id: 'w1', name: 'Report', workspaceId: 'ws-1' }])
+      queueTableRows(schemaMock.workspaceFiles, [])
+      mockDeduplicateWorkflowName.mockResolvedValueOnce('Report (2)')
+
+      await onBatch([{ id: 'folder-1' }])
+
+      // Deduped against the workspace ROOT (folderId null), which is where SET NULL would put it.
+      expect(mockDeduplicateWorkflowName).toHaveBeenCalledWith(
+        'Report',
+        'ws-1',
+        null,
+        expect.anything()
+      )
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({ folderId: null, name: 'Report (2)' })
+    })
+
+    it('re-roots an active workspace file under a deduplicated name', async () => {
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, [{ id: 'folder-1' }])
+      queueTableRows(schemaMock.workflow, [])
+      queueTableRows(schemaMock.workspaceFiles, [
+        { id: 'f1', originalName: 'report.pdf', workspaceId: 'ws-1' },
+      ])
+      mockAllocateUniqueWorkspaceFileName.mockResolvedValueOnce('report (2).pdf')
+
+      await onBatch([{ id: 'folder-1' }])
+
+      expect(mockAllocateUniqueWorkspaceFileName).toHaveBeenCalledWith('ws-1', 'report.pdf', null)
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({
+        folderId: null,
+        originalName: 'report (2).pdf',
+      })
+    })
+
+    it('falls back to an id-suffixed name when the copy-suffix range is exhausted', async () => {
+      // Letting the allocator throw would abort the sweep — the exact stall this guards against.
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, [{ id: 'folder-1' }])
+      queueTableRows(schemaMock.workflow, [])
+      queueTableRows(schemaMock.workspaceFiles, [
+        { id: 'f1', originalName: 'report.pdf', workspaceId: 'ws-1' },
+      ])
+      mockAllocateUniqueWorkspaceFileName.mockRejectedValueOnce(new Error('conflict'))
+
+      await expect(onBatch([{ id: 'folder-1' }])).resolves.toBeUndefined()
+
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({
+        folderId: null,
+        originalName: 'report.pdf (f1)',
+      })
+    })
+
+    it('re-roots an active SUBFOLDER, which hits the same unique index', async () => {
+      /**
+       * `folder.parentId` is also ON DELETE SET NULL and
+       * `folder_workspace_resource_parent_name_active_unique` keys on `coalesce(parent_id,'')`,
+       * so purging a parent can collide a surviving child at the root exactly like a workflow
+       * or file. Covering only those two would leave the class half-closed.
+       */
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, [{ id: 'folder-1' }])
+      queueTableRows(schemaMock.workflow, [])
+      queueTableRows(schemaMock.workspaceFiles, [])
+      queueTableRows(schemaMock.folder, [
+        { id: 'sub-1', name: 'Reports', workspaceId: 'ws-1', resourceType: 'knowledge_base' },
+      ])
+      mockDeduplicateFolderName.mockResolvedValueOnce('Reports (1)')
+
+      await onBatch([{ id: 'folder-1' }])
+
+      // Deduped against the ROOT of the child's OWN resourceType, which is where SET NULL lands it.
+      expect(mockDeduplicateFolderName).toHaveBeenCalledWith(
+        expect.anything(),
+        'ws-1',
+        null,
+        'Reports',
+        'knowledge_base'
+      )
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({ parentId: null, name: 'Reports (1)' })
+    })
+
+    it('recovers when the allocator RETURNS a colliding name and the update raises', async () => {
+      /**
+       * `allocateUniqueWorkspaceFileName` fails open — `fileExistsInWorkspace` swallows query
+       * errors and returns false — so it can hand back a name already taken at the root. Only
+       * the UPDATE discovers that, and an uncaught 23505 aborts the batch: the exact stall this
+       * hook prevents. Guarding the name lookup alone is not enough.
+       */
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, [{ id: 'folder-1' }])
+      queueTableRows(schemaMock.workflow, [])
+      queueTableRows(schemaMock.workspaceFiles, [
+        { id: 'f1', originalName: 'report.pdf', workspaceId: 'ws-1' },
+      ])
+      mockAllocateUniqueWorkspaceFileName.mockResolvedValueOnce('taken.pdf')
+      dbChainMockFns.update.mockImplementationOnce(() => ({
+        set: () => ({ where: () => Promise.reject(new Error('duplicate key value (23505)')) }),
+      }))
+
+      await expect(onBatch([{ id: 'folder-1' }])).resolves.toBeUndefined()
+
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({
+        folderId: null,
+        originalName: 'report.pdf (f1)',
+      })
+    })
+
+    it('leaves children alone when the folder was restored between select and onBatch', async () => {
+      /**
+       * The DELETE re-asserts eligibility and so correctly skips a restored folder. Without the
+       * same re-assertion here, this hook would still strip and rename that folder's children —
+       * leaving a live folder emptied out. This is the one side effect in the sweep that mutates
+       * rows which survive, so losing the race is user-visible.
+       */
+      const onBatch = await getFolderOnBatch()
+      queueTableRows(schemaMock.folder, []) // restored: no longer soft-deleted past retention
+      // Children ARE queued: without the eligibility re-check these would be re-rooted and
+      // renamed, so the assertions below fail rather than passing for want of candidate rows.
+      queueTableRows(schemaMock.workflow, [{ id: 'w1', name: 'Report', workspaceId: 'ws-1' }])
+      queueTableRows(schemaMock.workspaceFiles, [
+        { id: 'f1', originalName: 'report.pdf', workspaceId: 'ws-1' },
+      ])
+      dbChainMockFns.update.mockClear()
+
+      await onBatch([{ id: 'folder-1' }])
+
+      expect(mockDeduplicateWorkflowName).not.toHaveBeenCalled()
+      expect(mockAllocateUniqueWorkspaceFileName).not.toHaveBeenCalled()
+      expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    })
+
+    it('touches nothing when the batch is empty', async () => {
+      const onBatch = await getFolderOnBatch()
+      dbChainMockFns.select.mockClear()
+      dbChainMockFns.update.mockClear()
+
+      await onBatch([])
+
+      expect(dbChainMockFns.select).not.toHaveBeenCalled()
+      expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    })
+  })
 })

apps/sim/background/cleanup-soft-deletes.ts

@@ -29,10 +29,13 @@ import {
   selectRowsByIdChunks,
 } from '@/lib/cleanup/batch-delete'
 import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup'
+import { deduplicateFolderName } from '@/lib/folders/naming'
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
+import { allocateUniqueWorkspaceFileName } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
 import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
+import { deduplicateWorkflowName } from '@/lib/workflows/utils'
 
 const logger = createLogger('CleanupSoftDeletes')
 
@@ -424,6 +427,226 @@ async function cleanupExpiredKnowledgeBases(
  * workspace files → S3 storage) are handled explicitly so the SELECT that drives
  * the external cleanup and the SELECT that drives the DB delete see the same rows.
  */
+/** Per-run values an `onBatch` hook needs but `CLEANUP_TARGETS` cannot know at module scope. */
+interface CleanupBatchContext {
+  retentionDate: Date
+  label: string
+}
+
+/**
+ * Applies one re-root, treating the deduplicated name as a HINT rather than a guarantee.
+ *
+ * Both name allocators can hand back a name that is already taken: `fileExistsInWorkspace`
+ * swallows query errors and returns `false`, so `allocateUniqueWorkspaceFileName` fails OPEN,
+ * and `deduplicateWorkflowName`'s lookups can throw outright. Either way the UPDATE raises
+ * 23505, and an uncaught 23505 here aborts the batch — precisely the permanent retention stall
+ * this whole hook exists to prevent. So any failure retries with the row id, which is unique by
+ * construction and cannot collide.
+ *
+ * A failure of that retry is swallowed too: one unfixable row must not stop the other children
+ * from being made safe. It is logged at error level because the folder's DELETE can then still
+ * stall on that row via the FK's SET NULL.
+ */
+async function reRootOne(
+  preferred: () => Promise<unknown>,
+  withUniqueName: () => Promise<unknown>,
+  subject: string,
+  label: string
+): Promise<void> {
+  try {
+    await preferred()
+    return
+  } catch (error) {
+    logger.warn(`[${label}] Re-rooting ${subject} under its deduplicated name failed; retrying`, {
+      error,
+    })
+  }
+
+  try {
+    await withUniqueName()
+  } catch (error) {
+    logger.error(`[${label}] Could not re-root ${subject}; its folder delete may stall`, { error })
+  }
+}
+
+/**
+ * Re-roots any still-active workflow or workspace file filed under a folder that is about to be
+ * hard-deleted, giving it a collision-free name first.
+ *
+ * The `folder_id` FKs are `ON DELETE SET NULL`, so Postgres already re-roots these rows on its
+ * own. The problem is the name: both tables carry a partial unique index keyed on
+ * `coalesce(folder_id, '')`, so an implicit SET NULL can land a row on a name the workspace root
+ * already holds and abort the whole DELETE with a 23505. `chunkedBatchDelete` counts that as a
+ * failed batch and stops, and the same poison row re-fails on every later run — folder retention
+ * would stall permanently for that workspace chunk. Renaming here leaves the SET NULL a no-op.
+ *
+ * An active child inside a soft-deleted folder is already an anomaly — the delete cascade
+ * archives children — so this normally selects nothing, which is why the per-row loop is fine.
+ */
+async function reRootActiveFolderChildren(
+  folderIds: string[],
+  retentionDate: Date,
+  label: string
+): Promise<void> {
+  if (folderIds.length === 0) return
+  /**
+   * The SELECTs below are guarded for the same reason `reRootOne` swallows: this hook rejecting
+   * IS the aborted batch it exists to prevent. A transient read failure should leave the DELETE
+   * to succeed or fail on its own merits, not turn into a guaranteed stall.
+   */
+  try {
+    await reRootActiveFolderChildrenUnguarded(folderIds, retentionDate, label)
+  } catch (error) {
+    logger.error(`[${label}] Re-rooting children of purged folders failed`, { error })
+  }
+}
+
+async function reRootActiveFolderChildrenUnguarded(
+  folderIds: string[],
+  retentionDate: Date,
+  label: string
+): Promise<void> {
+  /**
+   * Re-asserted here, not just on the DELETE. `deleteFilter` already skips a folder restored
+   * between the SELECT and this hook — but without the same check here, this hook would still
+   * strip and rename the children of a folder that then survives, leaving a live folder
+   * emptied out. Every other side effect in this sweep only touches rows that are on their way
+   * out; this one mutates rows that stay, so it is the one place where losing that race is
+   * visible to the user.
+   *
+   * This narrows the window to match the DELETE's own re-assertion rather than closing it.
+   * Closing it properly means holding a row lock across select → onBatch → delete, which
+   * nothing in this sweep does today.
+   */
+  const stillExpired = await cleanupDb
+    .select({ id: folderTable.id })
+    .from(folderTable)
+    .where(
+      and(
+        inArray(folderTable.id, folderIds),
+        isNotNull(folderTable.deletedAt),
+        lt(folderTable.deletedAt, retentionDate)
+      )
+    )
+
+  const expiredIds = stillExpired.map(({ id }) => id)
+  if (expiredIds.length === 0) return
+
+  const workflows = await cleanupDb
+    .select({ id: workflow.id, name: workflow.name, workspaceId: workflow.workspaceId })
+    .from(workflow)
+    .where(and(inArray(workflow.folderId, expiredIds), isNull(workflow.archivedAt)))
+
+  for (const row of workflows) {
+    const workspaceId = row.workspaceId
+    if (!workspaceId) continue
+    await reRootOne(
+      async () => {
+        const name = await deduplicateWorkflowName(row.name, workspaceId, null, cleanupDb)
+        await cleanupDb
+          .update(workflow)
+          .set({ folderId: null, name })
+          .where(eq(workflow.id, row.id))
+      },
+      () =>
+        cleanupDb
+          .update(workflow)
+          .set({ folderId: null, name: `${row.name} (${row.id})` })
+          .where(eq(workflow.id, row.id)),
+      `workflow ${row.id}`,
+      label
+    )
+  }
+
+  const files = await cleanupDb
+    .select({
+      id: workspaceFiles.id,
+      originalName: workspaceFiles.originalName,
+      workspaceId: workspaceFiles.workspaceId,
+    })
+    .from(workspaceFiles)
+    .where(
+      and(
+        inArray(workspaceFiles.folderId, expiredIds),
+        isNull(workspaceFiles.deletedAt),
+        eq(workspaceFiles.context, 'workspace')
+      )
+    )
+
+  for (const row of files) {
+    const workspaceId = row.workspaceId
+    if (!workspaceId) continue
+    await reRootOne(
+      async () => {
+        const originalName = await allocateUniqueWorkspaceFileName(
+          workspaceId,
+          row.originalName,
+          null
+        )
+        await cleanupDb
+          .update(workspaceFiles)
+          .set({ folderId: null, originalName })
+          .where(eq(workspaceFiles.id, row.id))
+      },
+      () =>
+        cleanupDb
+          .update(workspaceFiles)
+          .set({ folderId: null, originalName: `${row.originalName} (${row.id})` })
+          .where(eq(workspaceFiles.id, row.id)),
+      `workspace file ${row.id}`,
+      label
+    )
+  }
+
+  /**
+   * Subfolders are exposed to exactly the same failure. `folder.parentId` is itself
+   * `ON DELETE SET NULL`, and `folder_workspace_resource_parent_name_active_unique` keys on
+   * `coalesce(parent_id, '')`, so purging a parent re-roots a surviving active child into a
+   * namespace where its name may already be taken — the identical 23505 stall. Covering only
+   * workflows and files would leave the class half-closed.
+   */
+  const childFolders = await cleanupDb
+    .select({
+      id: folderTable.id,
+      name: folderTable.name,
+      workspaceId: folderTable.workspaceId,
+      resourceType: folderTable.resourceType,
+    })
+    .from(folderTable)
+    .where(and(inArray(folderTable.parentId, expiredIds), isNull(folderTable.deletedAt)))
+
+  for (const row of childFolders) {
+    await reRootOne(
+      async () => {
+        const name = await deduplicateFolderName(
+          cleanupDb,
+          row.workspaceId,
+          null,
+          row.name,
+          row.resourceType
+        )
+        await cleanupDb
+          .update(folderTable)
+          .set({ parentId: null, name })
+          .where(eq(folderTable.id, row.id))
+      },
+      () =>
+        cleanupDb
+          .update(folderTable)
+          .set({ parentId: null, name: `${row.name} (${row.id})` })
+          .where(eq(folderTable.id, row.id)),
+      `folder ${row.id}`,
+      label
+    )
+  }
+
+  if (workflows.length > 0 || files.length > 0) {
+    logger.warn(
+      `[${label}] Re-rooted ${workflows.length} workflow(s) and ${files.length} file(s) out of folders being purged`
+    )
+  }
+}
+
 const CLEANUP_TARGETS = [
   {
     table: folderTable,
@@ -442,6 +665,12 @@ const CLEANUP_TARGETS = [
       'knowledge_base',
       'table',
     ]),
+    onBatch: (rows: { id: string }[], ctx: CleanupBatchContext) =>
+      reRootActiveFolderChildren(
+        rows.map(({ id }) => id),
+        ctx.retentionDate,
+        ctx.label
+      ),
     name: 'folder',
   },
   {
@@ -702,6 +931,10 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
       tableName: `${label}/${target.name}`,
       requireTimestampNotNull: true,
       additionalPredicate: 'additionalPredicate' in target ? target.additionalPredicate : undefined,
+      onBatch:
+        'onBatch' in target
+          ? (rows: { id: string }[]) => target.onBatch(rows, { retentionDate, label })
+          : undefined,
       dbClient: cleanupDb,
     })
     totalDeleted += result.deleted

apps/sim/lib/cleanup/batch-delete.test.ts

@@ -0,0 +1,80 @@
+/**
+ * @vitest-environment node
+ */
+
+import { schemaMock } from '@sim/testing'
+import { describe, expect, it, vi } from 'vitest'
+import { batchDeleteByWorkspaceAndTimestamp, chunkedBatchDelete } from '@/lib/cleanup/batch-delete'
+
+/**
+ * Minimal stand-in for the drizzle client `chunkedBatchDelete` calls. Only the DELETE path is
+ * modelled — the SELECT arrives through the caller-supplied `selectChunk`.
+ */
+function createDbClient(onDelete: () => void, selectRows: Array<{ id: string }> = []) {
+  return {
+    // `batchDeleteByWorkspaceAndTimestamp` builds its own selectChunk on this client.
+    select: () => ({
+      from: () => ({ where: () => ({ limit: async () => selectRows }) }),
+    }),
+    delete: () => {
+      onDelete()
+      return { where: () => ({ returning: async () => [{ id: 'row-1' }] }) }
+    },
+  } as never
+}
+
+/**
+ * These two assertions guard the single link that makes the folder re-root hook live. Every
+ * test in `cleanup-soft-deletes.test.ts` invokes the captured `onBatch` directly with
+ * `batchDeleteByWorkspaceAndTimestamp` mocked out, so a regression that stopped forwarding the
+ * hook — or ran it after the DELETE — would leave that whole suite green.
+ */
+describe('chunkedBatchDelete onBatch contract', () => {
+  it('runs onBatch BEFORE the delete for the same rows', async () => {
+    const order: string[] = []
+    const onBatch = vi.fn(async (rows: Array<{ id: string }>) => {
+      order.push(`onBatch:${rows.map((r) => r.id).join(',')}`)
+    })
+
+    await chunkedBatchDelete({
+      tableDef: schemaMock.folder as never,
+      workspaceIds: ['ws-1'],
+      tableName: 'test/folder',
+      dbClient: createDbClient(() => order.push('delete')),
+      selectChunk: async () => [{ id: 'row-1' }],
+      onBatch,
+      batchSize: 1,
+      maxBatches: 1,
+      totalRowLimit: 1,
+    })
+
+    expect(onBatch).toHaveBeenCalledWith([{ id: 'row-1' }])
+    // Ordering is the load-bearing half: renaming children after the DELETE would be useless.
+    expect(order).toEqual(['onBatch:row-1', 'delete'])
+  })
+
+  it('forwards onBatch through batchDeleteByWorkspaceAndTimestamp', async () => {
+    const order: string[] = []
+    const onBatch = vi.fn(async () => {
+      order.push('onBatch')
+    })
+
+    await batchDeleteByWorkspaceAndTimestamp({
+      tableDef: schemaMock.folder as never,
+      workspaceIdCol: schemaMock.folder.workspaceId as never,
+      timestampCol: schemaMock.folder.deletedAt as never,
+      workspaceIds: ['ws-1'],
+      retentionDate: new Date(0),
+      tableName: 'test/folder',
+      requireTimestampNotNull: true,
+      dbClient: createDbClient(() => order.push('delete'), [{ id: 'row-1' }]) as never,
+      onBatch,
+      batchSize: 1,
+      maxBatches: 1,
+    })
+
+    // The wrapper spreads `...rest` into chunkedBatchDelete; `onBatch` must survive that hop.
+    expect(onBatch).toHaveBeenCalled()
+    expect(order[0]).toBe('onBatch')
+  })
+})

apps/sim/lib/cleanup/batch-delete.ts

@@ -214,6 +214,12 @@ export interface BatchDeleteOptions {
    * so a cleanup pass only ever removes the kind it owns.
    */
   additionalPredicate?: SQL
+  /**
+   * Runs on each selected batch before its DELETE, for side effects that must observe exactly
+   * the rows about to be removed. Forwarded to `chunkedBatchDelete`; see `deleteFilter` there
+   * for the restore-race window this opens.
+   */
+  onBatch?: (rows: { id: string }[]) => Promise<void>
   batchSize?: number
   maxBatches?: number
   workspaceChunkSize?: number

apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts

@@ -267,7 +267,7 @@ function withCopySuffix(fileName: string, n: number): string {
 /**
  * Picks a display name that does not collide with an active workspace file (`original_name`).
  */
-async function allocateUniqueWorkspaceFileName(
+export async function allocateUniqueWorkspaceFileName(
   workspaceId: string,
   baseName: string,
   folderId?: string | null

packages/db/migrations/0276_drop_legacy_folder_tables.sql

@@ -0,0 +1,460 @@
+-- Contract migration for the generic-folders cutover: adopt the deferred `folder_id` foreign
+-- keys and drop the two legacy folder tables.
+--
+-- Ordering is deliberate and each step depends on the one before it:
+--   1. final insert-only reconcile, so no legacy folder is lost by the DROP — with one
+--      inherent exception: a legacy id already present in `folder` under a DIFFERENT
+--      resource_type cannot be inserted (the primary key is taken) and is dropped. That needs
+--      an id collision across two tables whose ids were preserved from disjoint sources, so it
+--      is not reachable in practice;
+--   2. re-root any `folder_id` that still does not resolve, so the FK can be validated;
+--   3. adopt the FKs the expand migration deliberately left off;
+--   4. drop the legacy tables.
+--
+-- Preconditions verified read-only against production before writing this file: 0 stranded
+-- rows in either tree, 0 unresolvable `folder_id`s, 0 active rows filed under a soft-deleted
+-- folder, and a FULL-ROW comparison (name, parent, deleted/archived state, workspace, user,
+-- sort order, locked) clean on both trees. The only divergence is 47 workflow-folder names, all
+-- matching `<old name> (N)` — 0272's deliberate dedup renames. That is precisely why step 1 is
+-- INSERT-ONLY: an upsert would revert all 47.
+--
+-- In production steps 1 and 2 are therefore no-ops. They exist for deployments that never ran
+-- the post-drain reconcile as an operational step — self-hosted upgrades above all, where a
+-- rolling restart can strand a folder exactly the same way and no operator is watching for it.
+-- This is the last moment the legacy rows exist, so it is the last chance to rescue them.
+--
+-- NAME DEDUPLICATION appears at four sites below and follows one pattern throughout, because
+-- three separate partial unique indexes are in play and every one of them keys on a coalesced
+-- nullable column, so re-rooting a row moves it into a namespace where its name may be taken:
+--   * folder            (workspace_id, resource_type, coalesce(parent_id,''), name) WHERE deleted_at IS NULL
+--   * workflow          (workspace_id, coalesce(folder_id,''), name)                WHERE archived_at IS NULL
+--   * workspace_files   (workspace_id, coalesce(folder_id,''), original_name)       WHERE deleted_at IS NULL AND context='workspace' AND workspace_id IS NOT NULL
+-- The pattern: rank contenders within the batch (`rn`), ask whether an already-present row
+-- holds the base name (`base_taken`), derive a `slot` into the free-suffix sequence, and probe
+-- for the first free `" (N)"` — where "free" must consider BOTH the rows already in the table
+-- AND the base names this batch is about to claim (`kept`). Probing only the table is the
+-- subtle failure: a stranded row legitimately named `Docs (1)` is invisible to the probe run
+-- for a stranded `Docs`, so both would be assigned `Docs (1)` and the statement aborts.
+-- Suffixes start at (1). Inactive rows are never renamed — the indexes are partial.
+
+-- Step 1 — final reconcile. Guarded on table existence so a replay after the DROP is a no-op
+-- rather than an error, and written as DO blocks so each tree is a single atomic statement
+-- (0272 ends with an embedded COMMIT, so this file is not guaranteed to run inside drizzle's
+-- batch transaction).
+DO $$
+BEGIN
+	IF to_regclass('public.workflow_folder') IS NULL THEN
+		RETURN;
+	END IF;
+
+	INSERT INTO "folder" (id, resource_type, name, user_id, workspace_id, parent_id, locked, sort_order, created_at, updated_at, deleted_at)
+	-- Keyed on `id` ALONE, matching the primary key it protects. Narrowing it by resource_type
+	-- would classify an id already present under a DIFFERENT type as stranded; the ON CONFLICT
+	-- below would then silently skip it rather than rescue it, so keeping the guard aligned with
+	-- the constraint is what makes the two agree.
+	WITH stranded AS (
+		SELECT l.id, l.name, l.user_id, l.workspace_id, l.parent_id, l.locked, l.sort_order,
+			l.created_at, l.updated_at, l.archived_at AS deleted_at
+		FROM "workflow_folder" l
+		WHERE NOT EXISTS (SELECT 1 FROM "folder" f WHERE f.id = l.id)
+	),
+	-- A parent is only usable if it will exist, shares this row's workspace, and leaves the row
+	-- REACHABLE. The workspace match is enforced by the `folder_parent_resource_type_match`
+	-- trigger and was never enforced by the legacy self-FK, so a cross-workspace parent is
+	-- representable in the source data. Reachability is the subtler half: filing an ACTIVE
+	-- folder under a soft-deleted parent hides it in Recently Deleted just as thoroughly as a
+	-- dangling parent would, so it re-roots too — matching `resolveRestoredFolderId`, which
+	-- re-roots a restored folder whose original parent is archived.
+	--
+	-- A soft-deleted row is exempt: it MAY keep a soft-deleted parent, because that is the
+	-- normal shape of an archived subtree and flattening it would destroy the hierarchy a
+	-- later restore rebuilds.
+	--
+	-- Anything else re-roots to the workspace root: losing one level of nesting beats losing the
+	-- folder and stranding every workflow inside it.
+	--
+	-- LIMITATION: this is a per-row check, so a CYCLE among stranded rows (a→b→a) survives it —
+	-- every row's parent exists, is same-workspace, and is active. Such rows land in `folder`
+	-- unreachable from the root. 0272's backfill has the identical hole, so this is not a
+	-- regression, and the client tolerates it (`getFolderPath` and `subtree.ts` both carry cycle
+	-- guards). Breaking cycles needs a recursive walk; it is deliberately not done here.
+	resolved AS (
+		SELECT s.*,
+			CASE
+				WHEN s.parent_id IS NULL THEN NULL
+				WHEN EXISTS (
+					SELECT 1 FROM "folder" f
+					WHERE f.id = s.parent_id AND f.resource_type = 'workflow' AND f.workspace_id = s.workspace_id
+						AND (s.deleted_at IS NOT NULL OR f.deleted_at IS NULL)
+				) THEN s.parent_id
+				WHEN EXISTS (
+					SELECT 1 FROM stranded s2
+					WHERE s2.id = s.parent_id AND s2.workspace_id = s.workspace_id
+						AND (s.deleted_at IS NOT NULL OR s2.deleted_at IS NULL)
+				) THEN s.parent_id
+				ELSE NULL
+			END AS resolved_parent
+		FROM stranded s
+	),
+	ranked AS (
+		SELECT r.*,
+			EXISTS (
+				SELECT 1 FROM "folder" a
+				WHERE a.workspace_id = r.workspace_id
+					AND a.resource_type = 'workflow'
+					AND coalesce(a.parent_id, '') = coalesce(r.resolved_parent, '')
+					AND a.name = r.name
+					AND a.deleted_at IS NULL
+			) AS base_taken,
+			row_number() OVER (
+				PARTITION BY r.workspace_id, coalesce(r.resolved_parent, ''), r.name
+				ORDER BY r.created_at, r.id
+			) AS rn
+		FROM resolved r
+		WHERE r.deleted_at IS NULL
+	),
+	slotted AS (
+		SELECT k.*, k.rn - 1 - (CASE WHEN k.base_taken THEN 0 ELSE 1 END) AS slot FROM ranked k
+	),
+	kept AS (
+		SELECT s.workspace_id, s.resolved_parent, s.name FROM slotted s WHERE s.slot < 0
+	),
+	named AS (
+		SELECT k.id,
+			CASE
+				WHEN k.slot < 0 THEN k.name
+				ELSE coalesce(
+					(
+						SELECT k.name || ' (' || candidate.n || ')'
+						FROM generate_series(1, 10000) AS candidate(n)
+						WHERE NOT EXISTS (
+							SELECT 1 FROM "folder" a
+							WHERE a.workspace_id = k.workspace_id
+								AND a.resource_type = 'workflow'
+								AND coalesce(a.parent_id, '') = coalesce(k.resolved_parent, '')
+								AND a.name = k.name || ' (' || candidate.n || ')'
+								AND a.deleted_at IS NULL
+						)
+						AND NOT EXISTS (
+							SELECT 1 FROM kept kp
+							WHERE kp.workspace_id = k.workspace_id
+								AND coalesce(kp.resolved_parent, '') = coalesce(k.resolved_parent, '')
+								AND kp.name = k.name || ' (' || candidate.n || ')'
+						)
+						ORDER BY candidate.n
+						OFFSET k.slot
+						LIMIT 1
+					),
+					-- Suffix space exhausted. Fall back to the id, which is unique by
+					-- construction, so the reconcile still completes and the row is traceable.
+					-- Falling back to the base name would guarantee a collision and abort here
+					-- with an error naming the base name, hiding the real cause.
+					k.name || ' (' || k.id || ')'
+				)
+			END AS final_name
+		FROM slotted k
+	)
+	SELECT r.id, 'workflow', coalesce(n.final_name, r.name),
+		r.user_id, r.workspace_id, r.resolved_parent, r.locked, r.sort_order,
+		r.created_at, r.updated_at, r.deleted_at
+	FROM resolved r
+	LEFT JOIN named n ON n.id = r.id
+	-- Matches the `stranded` guard, which already means "no folder row with this id". Restating
+	-- it as ON CONFLICT closes the gap between that read's snapshot and the index check: an
+	-- operational re-run of 0274, or a live pod, committing into `folder` mid-statement would
+	-- otherwise raise 23505 — and migrate.ts retries only 55P03, so that hard-fails the deploy.
+	ON CONFLICT (id) DO NOTHING;
+END $$;
+--> statement-breakpoint
+
+DO $$
+BEGIN
+	IF to_regclass('public.workspace_file_folders') IS NULL THEN
+		RETURN;
+	END IF;
+
+	INSERT INTO "folder" (id, resource_type, name, user_id, workspace_id, parent_id, locked, sort_order, created_at, updated_at, deleted_at)
+	WITH stranded AS (
+		SELECT l.id, l.name, l.user_id, l.workspace_id, l.parent_id, l.sort_order,
+			l.created_at, l.updated_at, l.deleted_at
+		FROM "workspace_file_folders" l
+		WHERE NOT EXISTS (SELECT 1 FROM "folder" f WHERE f.id = l.id)
+	),
+	-- Same reachability rule as the workflow tree above.
+	resolved AS (
+		SELECT s.*,
+			CASE
+				WHEN s.parent_id IS NULL THEN NULL
+				WHEN EXISTS (
+					SELECT 1 FROM "folder" f
+					WHERE f.id = s.parent_id AND f.resource_type = 'file' AND f.workspace_id = s.workspace_id
+						AND (s.deleted_at IS NOT NULL OR f.deleted_at IS NULL)
+				) THEN s.parent_id
+				WHEN EXISTS (
+					SELECT 1 FROM stranded s2
+					WHERE s2.id = s.parent_id AND s2.workspace_id = s.workspace_id
+						AND (s.deleted_at IS NOT NULL OR s2.deleted_at IS NULL)
+				) THEN s.parent_id
+				ELSE NULL
+			END AS resolved_parent
+		FROM stranded s
+	),
+	ranked AS (
+		SELECT r.*,
+			EXISTS (
+				SELECT 1 FROM "folder" a
+				WHERE a.workspace_id = r.workspace_id
+					AND a.resource_type = 'file'
+					AND coalesce(a.parent_id, '') = coalesce(r.resolved_parent, '')
+					AND a.name = r.name
+					AND a.deleted_at IS NULL
+			) AS base_taken,
+			row_number() OVER (
+				PARTITION BY r.workspace_id, coalesce(r.resolved_parent, ''), r.name
+				ORDER BY r.created_at, r.id
+			) AS rn
+		FROM resolved r
+		WHERE r.deleted_at IS NULL
+	),
+	slotted AS (
+		SELECT k.*, k.rn - 1 - (CASE WHEN k.base_taken THEN 0 ELSE 1 END) AS slot FROM ranked k
+	),
+	kept AS (
+		SELECT s.workspace_id, s.resolved_parent, s.name FROM slotted s WHERE s.slot < 0
+	),
+	named AS (
+		SELECT k.id,
+			CASE
+				WHEN k.slot < 0 THEN k.name
+				ELSE coalesce(
+					(
+						SELECT k.name || ' (' || candidate.n || ')'
+						FROM generate_series(1, 10000) AS candidate(n)
+						WHERE NOT EXISTS (
+							SELECT 1 FROM "folder" a
+							WHERE a.workspace_id = k.workspace_id
+								AND a.resource_type = 'file'
+								AND coalesce(a.parent_id, '') = coalesce(k.resolved_parent, '')
+								AND a.name = k.name || ' (' || candidate.n || ')'
+								AND a.deleted_at IS NULL
+						)
+						AND NOT EXISTS (
+							SELECT 1 FROM kept kp
+							WHERE kp.workspace_id = k.workspace_id
+								AND coalesce(kp.resolved_parent, '') = coalesce(k.resolved_parent, '')
+								AND kp.name = k.name || ' (' || candidate.n || ')'
+						)
+						ORDER BY candidate.n
+						OFFSET k.slot
+						LIMIT 1
+					),
+					k.name || ' (' || k.id || ')'
+				)
+			END AS final_name
+		FROM slotted k
+	)
+	SELECT r.id, 'file', coalesce(n.final_name, r.name),
+		r.user_id, r.workspace_id, r.resolved_parent, false, r.sort_order,
+		r.created_at, r.updated_at, r.deleted_at
+	FROM resolved r
+	LEFT JOIN named n ON n.id = r.id
+	-- Same rationale as the workflow tree above.
+	ON CONFLICT (id) DO NOTHING;
+END $$;
+--> statement-breakpoint
+
+-- Step 2 — a `folder_id` can only still dangle if its folder is absent from BOTH tables, which
+-- step 1 cannot rescue. Re-root it so the resource stays reachable at the workspace root
+-- instead of blocking validation, and rename on collision for the same reason step 1 does: the
+-- root namespace is covered by a partial unique index, and two same-named rows re-rooted out of
+-- two different vanished folders would abort the migration. A dangling row is already
+-- unreachable in the UI (filed under a folder that does not exist), so surfacing it at the root
+-- under a suffixed name strictly improves on leaving it invisible.
+DO $$
+BEGIN
+	-- `workspace_id IS NOT NULL` is load-bearing, not defensive. `workflow.workspace_id` is
+	-- nullable (personal workflows), and NULL is treated as EQUAL by `PARTITION BY` but as
+	-- UNKNOWN by the `=` in `base_taken`. Without this guard `rn` increments across every
+	-- personal workflow while no collision is ever detected, so the dedup can ONLY fire
+	-- spuriously — renaming a user-visible workflow that needed no rename, since the unique
+	-- index treats NULL `workspace_id` rows as distinct anyway. The file block below has always
+	-- carried the equivalent guard.
+	WITH dangling AS (
+		SELECT w.id, w.workspace_id, w.name,
+			(w.archived_at IS NULL AND w.workspace_id IS NOT NULL) AS is_active
+		FROM "workflow" w
+		WHERE w."folder_id" IS NOT NULL
+			AND NOT EXISTS (SELECT 1 FROM "folder" f WHERE f.id = w."folder_id")
+	),
+	ranked AS (
+		SELECT d.*,
+			EXISTS (
+				SELECT 1 FROM "workflow" r
+				WHERE r.workspace_id = d.workspace_id
+					AND r."folder_id" IS NULL
+					AND r.archived_at IS NULL
+					AND r.name = d.name
+			) AS base_taken,
+			row_number() OVER (PARTITION BY d.workspace_id, d.name ORDER BY d.id) AS rn
+		FROM dangling d
+		WHERE d.is_active
+	),
+	slotted AS (
+		SELECT k.*, k.rn - 1 - (CASE WHEN k.base_taken THEN 0 ELSE 1 END) AS slot FROM ranked k
+	),
+	kept AS (
+		SELECT s.workspace_id, s.name FROM slotted s WHERE s.slot < 0
+	),
+	named AS (
+		SELECT k.id,
+			CASE
+				WHEN k.slot < 0 THEN k.name
+				ELSE coalesce(
+					(
+						SELECT k.name || ' (' || candidate.n || ')'
+						FROM generate_series(1, 10000) AS candidate(n)
+						WHERE NOT EXISTS (
+							SELECT 1 FROM "workflow" r
+							WHERE r.workspace_id = k.workspace_id
+								AND r."folder_id" IS NULL
+								AND r.archived_at IS NULL
+								AND r.name = k.name || ' (' || candidate.n || ')'
+						)
+						AND NOT EXISTS (
+							SELECT 1 FROM kept kp
+							WHERE kp.workspace_id = k.workspace_id
+								AND kp.name = k.name || ' (' || candidate.n || ')'
+						)
+						ORDER BY candidate.n
+						OFFSET k.slot
+						LIMIT 1
+					),
+					k.name || ' (' || k.id || ')'
+				)
+			END AS final_name
+		FROM slotted k
+	)
+	UPDATE "workflow" w
+	SET "folder_id" = NULL, "name" = coalesce(n.final_name, w.name)
+	FROM dangling d
+	LEFT JOIN named n ON n.id = d.id
+	WHERE w.id = d.id;
+END $$;
+--> statement-breakpoint
+
+DO $$
+BEGIN
+	WITH dangling AS (
+		SELECT wf.id, wf.workspace_id, wf.original_name,
+			(wf.deleted_at IS NULL AND wf.context = 'workspace' AND wf.workspace_id IS NOT NULL) AS is_active
+		FROM "workspace_files" wf
+		WHERE wf."folder_id" IS NOT NULL
+			AND NOT EXISTS (SELECT 1 FROM "folder" f WHERE f.id = wf."folder_id")
+	),
+	ranked AS (
+		SELECT d.*,
+			EXISTS (
+				SELECT 1 FROM "workspace_files" r
+				WHERE r.workspace_id = d.workspace_id
+					AND r."folder_id" IS NULL
+					AND r.deleted_at IS NULL
+					AND r.context = 'workspace'
+					AND r.original_name = d.original_name
+			) AS base_taken,
+			row_number() OVER (PARTITION BY d.workspace_id, d.original_name ORDER BY d.id) AS rn
+		FROM dangling d
+		WHERE d.is_active
+	),
+	slotted AS (
+		SELECT k.*, k.rn - 1 - (CASE WHEN k.base_taken THEN 0 ELSE 1 END) AS slot FROM ranked k
+	),
+	kept AS (
+		SELECT s.workspace_id, s.original_name FROM slotted s WHERE s.slot < 0
+	),
+	named AS (
+		SELECT k.id,
+			CASE
+				WHEN k.slot < 0 THEN k.original_name
+				ELSE coalesce(
+					(
+						SELECT k.original_name || ' (' || candidate.n || ')'
+						FROM generate_series(1, 10000) AS candidate(n)
+						WHERE NOT EXISTS (
+							SELECT 1 FROM "workspace_files" r
+							WHERE r.workspace_id = k.workspace_id
+								AND r."folder_id" IS NULL
+								AND r.deleted_at IS NULL
+								AND r.context = 'workspace'
+								AND r.original_name = k.original_name || ' (' || candidate.n || ')'
+						)
+						AND NOT EXISTS (
+							SELECT 1 FROM kept kp
+							WHERE kp.workspace_id = k.workspace_id
+								AND kp.original_name = k.original_name || ' (' || candidate.n || ')'
+						)
+						ORDER BY candidate.n
+						OFFSET k.slot
+						LIMIT 1
+					),
+					k.original_name || ' (' || k.id || ')'
+				)
+			END AS final_name
+		FROM slotted k
+	)
+	UPDATE "workspace_files" wf
+	SET "folder_id" = NULL, "original_name" = coalesce(n.final_name, wf.original_name)
+	FROM dangling d
+	LEFT JOIN named n ON n.id = d.id
+	WHERE wf.id = d.id;
+END $$;
+--> statement-breakpoint
+
+-- Step 3 — adopt the FKs 0272 deliberately left off. Added NOT VALID so the ACCESS EXCLUSIVE
+-- lock covers only the catalog write, not a full scan: `workspace_files` is ~1.7M rows / 1.8GB
+-- and an immediately-validated FK would block every read and write on it for the whole scan.
+-- NOT VALID still enforces the constraint on all new writes; VALIDATE below takes only SHARE
+-- UPDATE EXCLUSIVE and so runs concurrently with normal traffic.
+DO $$
+BEGIN
+	ALTER TABLE "workflow"
+		ADD CONSTRAINT "workflow_folder_id_folder_id_fk"
+		FOREIGN KEY ("folder_id") REFERENCES "public"."folder"("id") ON DELETE SET NULL NOT VALID;
+EXCEPTION
+	WHEN duplicate_object THEN NULL;
+END $$;
+--> statement-breakpoint
+
+DO $$
+BEGIN
+	ALTER TABLE "workspace_files"
+		ADD CONSTRAINT "workspace_files_folder_id_folder_id_fk"
+		FOREIGN KEY ("folder_id") REFERENCES "public"."folder"("id") ON DELETE SET NULL NOT VALID;
+EXCEPTION
+	WHEN duplicate_object THEN NULL;
+END $$;
+--> statement-breakpoint
+
+-- The reconcile and the NOT VALID constraints must be durable before the scans below, which
+-- deliberately run outside the surrounding transaction so they do not hold its locks.
+COMMIT;
+--> statement-breakpoint
+
+ALTER TABLE "workflow" VALIDATE CONSTRAINT "workflow_folder_id_folder_id_fk";
+--> statement-breakpoint
+ALTER TABLE "workspace_files" VALIDATE CONSTRAINT "workspace_files_folder_id_folder_id_fk";
+--> statement-breakpoint
+
+-- Step 4 — drop the legacy tables. Named EXACTLY and never by pattern: `workflow_folder_sort_idx`
+-- is an index on the LIVE `workflow` table, so anything globbing `workflow_folder*` would take
+-- out a production index. Their own FKs and indexes go with them; nothing references either
+-- table, so no CASCADE is needed and its absence is the safety check.
+--
+-- The cutover that stopped all reads and writes of these two tables shipped in EARLIER deploys
+-- (#6037 / #6045), not in this PR. Production has since drained — last legacy write 06:27:21Z,
+-- verified >10h earlier — and the full-row comparison described at the top of this file confirms
+-- nothing is stranded. Step 1 rescues any straggler regardless.
+-- migration-safe: reads/writes ceased in an earlier deploy; drained and full-row verified.
+DROP TABLE IF EXISTS "workflow_folder";
+--> statement-breakpoint
+-- migration-safe: same cutover, same drain, same full-row verification as the drop above.
+DROP TABLE IF EXISTS "workspace_file_folders";

packages/db/migrations/meta/_journal.json

@@ -1926,6 +1926,13 @@
       "when": 1785344855092,
       "tag": "0275_table_views",
       "breakpoints": true
+    },
+    {
+      "idx": 276,
+      "version": "7",
+      "when": 1785352177983,
+      "tag": "0276_drop_legacy_folder_tables",
+      "breakpoints": true
     }
   ]
 }

packages/db/schema.ts

@@ -124,8 +124,8 @@ export const folderResourceTypeEnum = pgEnum('folder_resource_type', [
 
 /**
  * Generic folder hierarchy shared by workflows, files, knowledge bases, and tables.
- * Supersedes the resource-specific `workflowFolder`/`workspaceFileFolder` tables (see
- * the deprecation notes on those below).
+ * Supersedes the resource-specific `workflow_folder` and `workspace_file_folders` tables,
+ * dropped in migration 0276 once the cutover was verified against production.
  *
  * `resourceType` is a real `pgEnum` here — unlike `pinnedItem.resourceType` — because the
  * set of folder-bearing resources is small and fixed. A folder may only parent a folder
@@ -138,7 +138,8 @@ export const folderResourceTypeEnum = pgEnum('folder_resource_type', [
  * `false` and no lock cascade reads it for them. Dropping the column would regress
  * shipped workflow-folder locking.
  *
- * `color` and `isExpanded` from `workflowFolder` are intentionally not carried over:
+ * `color` and `isExpanded` from the old `workflow_folder` table are intentionally not
+ * carried over:
  * `color` has no UI consumer, and `isExpanded`'s real state lives client-side in the
  * folders Zustand store and is never read back from the DB.
  */
@@ -176,9 +177,9 @@ export const folder = pgTable(
       .on(table.workspaceId, table.deletedAt)
       .where(sql`${table.deletedAt} IS NOT NULL`),
     /**
-     * Mirrors `workspace_file_folders_workspace_parent_name_active_unique`, which file
-     * folders already enforce today. Workflow folders gain it here — the backfill
-     * deduplicates the existing violations.
+     * Carries over the active-unique key the old `workspace_file_folders` table enforced,
+     * and extends it to workflow folders, which never had one — 0272's backfill deduplicated
+     * the 47 pre-existing violations it surfaced.
      */
     workspaceResourceParentNameActiveUnique: uniqueIndex(
       'folder_workspace_resource_parent_name_active_unique'
@@ -224,43 +225,6 @@ export const pinnedItem = pgTable(
   })
 )
 
-// DEPRECATED: superseded by the generic `folder` table (resourceType='workflow'). Kept
-// (unread, unwritten) until the generic-folders cutover is verified in production; dropped
-// in a follow-up contract migration.
-export const workflowFolder = pgTable(
-  'workflow_folder',
-  {
-    id: text('id').primaryKey(),
-    name: text('name').notNull(),
-    userId: text('user_id')
-      .notNull()
-      .references(() => user.id, { onDelete: 'cascade' }),
-    workspaceId: text('workspace_id')
-      .notNull()
-      .references(() => workspace.id, { onDelete: 'cascade' }),
-    parentId: text('parent_id'), // Self-reference will be handled by foreign key constraint
-    color: text('color').default('#6B7280'),
-    isExpanded: boolean('is_expanded').notNull().default(true),
-    locked: boolean('locked').notNull().default(false),
-    sortOrder: integer('sort_order').notNull().default(0),
-    createdAt: timestamp('created_at').notNull().defaultNow(),
-    updatedAt: timestamp('updated_at').notNull().defaultNow(),
-    archivedAt: timestamp('archived_at'),
-  },
-  (table) => ({
-    userIdx: index('workflow_folder_user_idx').on(table.userId),
-    workspaceParentIdx: index('workflow_folder_workspace_parent_idx').on(
-      table.workspaceId,
-      table.parentId
-    ),
-    parentSortIdx: index('workflow_folder_parent_sort_idx').on(table.parentId, table.sortOrder),
-    archivedAtIdx: index('workflow_folder_archived_at_idx').on(table.archivedAt),
-    workspaceArchivedAtPartialIdx: index('workflow_folder_workspace_archived_partial_idx')
-      .on(table.workspaceId, table.archivedAt)
-      .where(sql`${table.archivedAt} IS NOT NULL`),
-  })
-)
-
 export const workflow = pgTable(
   'workflow',
   {
@@ -269,15 +233,7 @@ export const workflow = pgTable(
       .notNull()
       .references(() => user.id, { onDelete: 'cascade' }),
     workspaceId: text('workspace_id').references(() => workspace.id, { onDelete: 'cascade' }),
-    /**
-     * contract-pending: re-add `.references(() => folder.id, { onDelete: 'set null' })`
-     * once the generic-folders expand migration is fully deployed and no old-code pod can
-     * still write a workflow_folder-only id here. The expand migration only DROPs the old
-     * (now-wrong) FK target; adding the new one in the same deploy would reject writes from
-     * still-running old app code. The invariant is enforced in the application layer
-     * meanwhile.
-     */
-    folderId: text('folder_id'),
+    folderId: text('folder_id').references(() => folder.id, { onDelete: 'set null' }),
     sortOrder: integer('sort_order').notNull().default(0),
     name: text('name').notNull(),
     description: text('description'),
@@ -1902,62 +1858,6 @@ export const workspaceFile = pgTable(
   })
 )
 
-/**
- * DEPRECATED: superseded by the generic `folder` table (`resource_type = 'file'`).
- *
- * As of the file-folder cutover the application no longer reads or writes this table —
- * every former query site now targets `folder` scoped to `resource_type = 'file'`. It is
- * retained as the rollback copy for that deploy, NOT because it is already unused history:
- * before the cutover it was the live table, and migration 0272's one-shot backfill did not
- * cover folders created after it ran (migration 0274 catches those up).
- *
- * Do NOT drop it in a contract migration until BOTH hold: the cutover has been running in
- * production long enough that a rollback is off the table, and a FULL-ROW comparison against
- * `folder WHERE resource_type = 'file'` confirms nothing was stranded. A row COUNT is not
- * sufficient — the pre-0274 divergence was in row contents (names, parents, `deleted_at`),
- * which a count check passes straight through. Dropping this on the strength of "no code
- * references it" alone destroys the only rollback path.
- */
-export const workspaceFileFolder = pgTable(
-  'workspace_file_folders',
-  {
-    id: text('id').primaryKey(),
-    name: text('name').notNull(),
-    userId: text('user_id')
-      .notNull()
-      .references(() => user.id, { onDelete: 'cascade' }),
-    workspaceId: text('workspace_id')
-      .notNull()
-      .references(() => workspace.id, { onDelete: 'cascade' }),
-    parentId: text('parent_id').references((): AnyPgColumn => workspaceFileFolder.id, {
-      onDelete: 'set null',
-    }),
-    sortOrder: integer('sort_order').notNull().default(0),
-    deletedAt: timestamp('deleted_at'),
-    createdAt: timestamp('created_at').notNull().defaultNow(),
-    updatedAt: timestamp('updated_at').notNull().defaultNow(),
-  },
-  (table) => ({
-    workspaceParentIdx: index('workspace_file_folders_workspace_parent_idx').on(
-      table.workspaceId,
-      table.parentId
-    ),
-    parentSortIdx: index('workspace_file_folders_parent_sort_idx').on(
-      table.parentId,
-      table.sortOrder
-    ),
-    deletedAtIdx: index('workspace_file_folders_deleted_at_idx').on(table.deletedAt),
-    workspaceDeletedAtPartialIdx: index('workspace_file_folders_workspace_deleted_partial_idx')
-      .on(table.workspaceId, table.deletedAt)
-      .where(sql`${table.deletedAt} IS NOT NULL`),
-    workspaceParentNameActiveUnique: uniqueIndex(
-      'workspace_file_folders_workspace_parent_name_active_unique'
-    )
-      .on(table.workspaceId, sql`coalesce(${table.parentId}, '')`, table.name)
-      .where(sql`${table.deletedAt} IS NULL`),
-  })
-)
-
 export const workspaceFiles = pgTable(
   'workspace_files',
   {
@@ -1967,15 +1867,7 @@ export const workspaceFiles = pgTable(
       .notNull()
       .references(() => user.id, { onDelete: 'cascade' }),
     workspaceId: text('workspace_id').references(() => workspace.id, { onDelete: 'cascade' }),
-    /**
-     * contract-pending: re-add `.references(() => folder.id, { onDelete: 'set null' })`
-     * once the generic-folders expand migration is fully deployed and no old-code pod can
-     * still write a workspace_file_folders-only id here. The expand migration only DROPs the old
-     * (now-wrong) FK target; adding the new one in the same deploy would reject writes from
-     * still-running old app code. The invariant is enforced in the application layer
-     * meanwhile.
-     */
-    folderId: text('folder_id'),
+    folderId: text('folder_id').references(() => folder.id, { onDelete: 'set null' }),
     context: text('context').notNull(), // 'workspace', 'mothership', 'copilot', 'chat', 'knowledge-base', 'profile-pictures', 'general', 'execution'
     chatId: uuid('chat_id').references(() => copilotChats.id, { onDelete: 'cascade' }),
     /**

packages/testing/src/mocks/schema.mock.ts

@@ -89,19 +89,6 @@ export const schemaMock = {
     resourceId: 'resourceId',
     pinnedAt: 'pinnedAt',
   },
-  workflowFolder: {
-    id: 'id',
-    name: 'name',
-    userId: 'userId',
-    workspaceId: 'workspaceId',
-    parentId: 'parentId',
-    color: 'color',
-    isExpanded: 'isExpanded',
-    sortOrder: 'sortOrder',
-    createdAt: 'createdAt',
-    updatedAt: 'updatedAt',
-    archivedAt: 'archivedAt',
-  },
   workflow: {
     id: 'id',
     userId: 'userId',
@@ -557,17 +544,6 @@ export const schemaMock = {
     completedAt: 'completedAt',
     updatedAt: 'updatedAt',
   },
-  workspaceFileFolder: {
-    id: 'id',
-    name: 'name',
-    userId: 'userId',
-    workspaceId: 'workspaceId',
-    parentId: 'parentId',
-    sortOrder: 'sortOrder',
-    deletedAt: 'deletedAt',
-    createdAt: 'createdAt',
-    updatedAt: 'updatedAt',
-  },
   workspaceFile: {
     id: 'id',
     workspaceId: 'workspaceId',