Merge remote-tracking branch 'origin/staging' into codex/durable-agent-memory

Author: icecrasher321

apps/docs/openapi-v2-files-audit.json

@@ -868,7 +868,7 @@
       "get": {
         "operationId": "listFileVersions",
         "summary": "List File Versions",
-        "description": "List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits within ten minutes fold into one version, as do repeated workflow writes by one author. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten and removes older versions by plan, so numbers can have gaps.\n\nOAuth scope: `api:read`.",
+        "description": "List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.",
         "x-sim-operation": "files.versions.list",
         "x-oauth-scope": "api:read",
         "tags": ["Files"],

apps/sim/background/cleanup-file-versions.ts

@@ -6,6 +6,7 @@ import { task } from '@trigger.dev/sdk'
 import { and, count, gt, inArray, isNotNull, lt, min, or, sql } from 'drizzle-orm'
 import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
 import {
+  DEFAULT_BATCH_SIZE,
   DEFAULT_DELETE_CHUNK_SIZE,
   DEFAULT_MAX_BATCHES_PER_TABLE,
   DEFAULT_WORKSPACE_CHUNK_SIZE,
@@ -22,6 +23,12 @@ const cleanupDb = dbFor('cleanup')
 /** Candidate files whose histories are ranked in one query. */
 const FILES_PER_QUERY = 500
 
+/**
+ * Bounds one run like the other cleanup jobs: {@link DEFAULT_MAX_BATCHES_PER_TABLE} batches per
+ * workspace chunk and this many versions overall. The next run resumes where this one stopped.
+ */
+const MAX_VERSIONS_PER_RUN = DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE
+
 /**
  * Superseded versions a free file keeps (its newest 100 with the current one); versions beyond it
  * are pruned whatever their age. Paid plans are bounded only by the inline write-time ceiling.
@@ -73,7 +80,12 @@ async function selectCandidateFileIds(
  * Superseded versions of the given files past retention: older than the cutoff or beyond the plan's
  * count, but never among the newest {@link KEEP_SUPERSEDED} superseded versions of a file.
  */
-function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: number) {
+function selectExpiredVersions(
+  fileIds: string[],
+  cutoff: Date,
+  maxSuperseded: number,
+  batchSize: number
+) {
   const ranked = cleanupDb
     .select({
       id: workspaceFileVersion.id,
@@ -100,7 +112,7 @@ function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: n
         or(lt(ranked.supersededAt, cutoff), gt(ranked.rank, maxSuperseded))
       )
     )
-    .limit(DEFAULT_DELETE_CHUNK_SIZE)
+    .limit(batchSize)
 }
 
 /**
@@ -146,16 +158,27 @@ export async function runCleanupFileVersions(payload: CleanupJobPayload): Promis
   )
 
   let deleted = 0
+  let attempted = 0
   for (const group of chunkArray(workspaceIds, DEFAULT_WORKSPACE_CHUNK_SIZE)) {
+    if (attempted >= MAX_VERSIONS_PER_RUN) break
     const candidates = await selectCandidateFileIds(group, cutoff, maxSuperseded)
+    let batches = 0
     for (const fileIds of chunkArray(candidates, FILES_PER_QUERY)) {
-      for (let batch = 0; batch < DEFAULT_MAX_BATCHES_PER_TABLE; batch++) {
-        const expired = await selectExpiredVersions(fileIds, cutoff, maxSuperseded)
-        if (expired.length === 0) break
-        const removed = await deleteVersions(expired)
+      let exhausted = false
+      while (
+        !exhausted &&
+        batches < DEFAULT_MAX_BATCHES_PER_TABLE &&
+        attempted < MAX_VERSIONS_PER_RUN
+      ) {
+        batches++
+        const batchSize = Math.min(DEFAULT_DELETE_CHUNK_SIZE, MAX_VERSIONS_PER_RUN - attempted)
+        const expired = await selectExpiredVersions(fileIds, cutoff, maxSuperseded, batchSize)
+        attempted += expired.length
+        const removed = expired.length > 0 ? await deleteVersions(expired) : 0
         deleted += removed
-        if (expired.length < DEFAULT_DELETE_CHUNK_SIZE || removed === 0) break
+        exhausted = expired.length < batchSize || removed === 0
       }
+      if (!exhausted) break
     }
   }
 

apps/sim/lib/api/contracts/v2/openapi/files-audit.ts

@@ -447,7 +447,7 @@ const declaredRoutes = [
       operationId: 'listFileVersions',
       summary: 'List File Versions',
       description:
-        'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits within ten minutes fold into one version, as do repeated workflow writes by one author. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten and removes older versions by plan, so numbers can have gaps.',
+        'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.',
       errors: RESOURCE_ERRORS,
       success: { description: 'A page of file versions.' },
     }),

apps/sim/lib/api/mcp/generated/v2-operations.ts

@@ -1425,7 +1425,7 @@ export const V2_MCP_OPERATIONS = {
     contract: v2ListFileVersionsContract,
     summary: 'List File Versions',
     description:
-      'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits within ten minutes fold into one version, as do repeated workflow writes by one author. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten and removes older versions by plan, so numbers can have gaps.\n\nOAuth scope: `api:read`.',
+      'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.',
     handler: () => import('@/app/api/v2/files/[fileId]/versions/route').then((route) => route.GET),
   },
   listKnowledgeBases: {

apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts

@@ -1,6 +1,6 @@
 /** Real PostgreSQL transactions and local object storage for workspace file version history. */
 import { mkdtempSync } from 'node:fs'
-import { access, rm } from 'node:fs/promises'
+import { access, mkdir, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import path from 'node:path'
 import { db, dbFor } from '@sim/db'
@@ -39,6 +39,7 @@ import {
 import { WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
 import {
   getCurrentWorkspaceFileVersion,
+  getWorkspaceFileVersion,
   queryWorkspaceFileVersions,
   releaseWorkspaceFileVersionsForPurgeInTx,
 } from '@/lib/uploads/contexts/workspace/workspace-file-versions'
@@ -288,16 +289,123 @@ describe('workspace file version history in PostgreSQL', () => {
     )
 
     await expect(deleteWorkspaceFileVersion(fixture.workspaceId, fixture.fileId, 2)).resolves.toBe(
-      false
+      'newest'
     )
     await expect(deleteWorkspaceFileVersion(fixture.workspaceId, fixture.fileId, 1)).resolves.toBe(
-      true
+      'deleted'
+    )
+    await expect(deleteWorkspaceFileVersion(fixture.workspaceId, fixture.fileId, 1)).resolves.toBe(
+      'not_found'
     )
 
     expect((await versionRows(fixture.fileId)).map((row) => row.version)).toEqual([2])
     expect(await objectExists(fixture.firstKey)).toBe(false)
   })
 
+  it('reads bytes a write left unrecorded as the version its next write records them as', async () => {
+    const fixture = await seedFile('original')
+    const write = { source: 'api', authorUserId: fixture.aliceId } as const
+    await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.aliceId,
+      Buffer.from('second'),
+      undefined,
+      { version: write }
+    )
+    /** A content write that replaced the bytes without recording a version, as a build predating history would. */
+    const unrecordedKey = `${fixture.firstKey}-unrecorded`
+    const unrecordedContent = 'third, never recorded'
+    const unrecordedPath = path.join(fixtureStorage.root, unrecordedKey)
+    await mkdir(path.dirname(unrecordedPath), { recursive: true })
+    await writeFile(unrecordedPath, unrecordedContent)
+    await db
+      .update(workspaceFiles)
+      .set({
+        key: unrecordedKey,
+        sizeBytes: Buffer.byteLength(unrecordedContent),
+        contentUpdatedAt: new Date(),
+      })
+      .where(eq(workspaceFiles.id, fixture.fileId))
+    const file = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+    if (!file) throw new Error('file missing')
+
+    await expect(
+      getWorkspaceFileWithCurrentVersion(fixture.workspaceId, fixture.fileId)
+    ).resolves.toMatchObject({ key: unrecordedKey, currentVersion: 3 })
+    expect(await getCurrentWorkspaceFileVersion(file)).toMatchObject({
+      version: 3,
+      key: unrecordedKey,
+      isCurrent: true,
+    })
+    const listed = await queryWorkspaceFileVersions(file, { sortOrder: 'desc', limit: 10 })
+    expect(listed.versions[0].size).toBe(Buffer.byteLength(unrecordedContent))
+    expect(listed.versions.map((row) => [row.version, row.isCurrent, row.source])).toEqual([
+      [3, true, 'unknown'],
+      [2, false, 'api'],
+      [1, false, 'upload'],
+    ])
+    expect(listed.versions[1].supersededAt).toEqual(file.contentUpdatedAt)
+
+    const firstPage = await queryWorkspaceFileVersions(file, { sortOrder: 'asc', limit: 2 })
+    expect(firstPage.versions.map((row) => row.version)).toEqual([1, 2])
+    const lastPage = await queryWorkspaceFileVersions(file, {
+      sortOrder: 'asc',
+      limit: 2,
+      after: firstPage.nextKeys ?? undefined,
+    })
+    expect(lastPage.versions.map((row) => row.version)).toEqual([3])
+    expect(lastPage.nextKeys).toBeNull()
+
+    await expect(deleteWorkspaceFileVersion(fixture.workspaceId, fixture.fileId, 2)).resolves.toBe(
+      'newest'
+    )
+
+    const next = await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.aliceId,
+      Buffer.from('fourth'),
+      undefined,
+      { version: write }
+    )
+    expect(next.currentVersion).toBe(4)
+    expect((await versionRows(fixture.fileId)).map((row) => [row.version, row.source])).toEqual([
+      [1, 'upload'],
+      [2, 'api'],
+      [3, 'unknown'],
+      [4, 'api'],
+    ])
+    const materialized = (await versionRows(fixture.fileId))[2]
+    expect(materialized.key).toBe(unrecordedKey)
+    expect(materialized.sizeBytes).toBe(Buffer.byteLength(unrecordedContent))
+    expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, materialized.key)).toBe(
+      unrecordedContent
+    )
+  })
+
+  it('reads versions against the file as committed, not a record loaded before a write', async () => {
+    const fixture = await seedFile('original')
+    const stale = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+    if (!stale) throw new Error('file missing')
+    await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.aliceId,
+      Buffer.from('second'),
+      undefined,
+      { version: { source: 'api', authorUserId: fixture.aliceId } }
+    )
+
+    const listed = await queryWorkspaceFileVersions(stale, { sortOrder: 'desc', limit: 10 })
+    expect(listed.versions.map((row) => [row.version, row.isCurrent])).toEqual([
+      [2, true],
+      [1, false],
+    ])
+    expect((await getCurrentWorkspaceFileVersion(stale)).version).toBe(2)
+    await expect(getWorkspaceFileVersion(stale, 3)).resolves.toBeNull()
+  })
+
   it('never folds deliberate writes, and repoints the head for identical bytes', async () => {
     const fixture = await seedFile('original')
     const write = { source: 'api', authorUserId: fixture.aliceId } as const

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

@@ -82,6 +82,7 @@ import {
   listWorkspaceFileVersionKeysInTx,
   loadWorkspaceFileVersionHead,
   recordWorkspaceFileVersionInTx,
+  type WorkspaceFileVersionDeletion,
   type WorkspaceFileVersionWrite,
 } from '@/lib/uploads/contexts/workspace/workspace-file-versions'
 import { buildStorageKeySegment } from '@/lib/uploads/core/storage-key'
@@ -2097,33 +2098,35 @@ export async function updateWorkspaceFileContent(
 /**
  * Deletes one superseded version of an active workspace file and releases its stored object. The
  * file row is locked so the delete serializes with content writes that supersede or prune history.
- * Returns false when the file has no such superseded version (it never existed, retention removed
- * it, or it is the current version).
  */
 export async function deleteWorkspaceFileVersion(
   workspaceId: string,
   fileId: string,
   version: number
-): Promise<boolean> {
-  const cleanupEventIds = await db.transaction(async (tx) => {
+): Promise<WorkspaceFileVersionDeletion['status']> {
+  const deletion = await db.transaction(async (tx) => {
     const [file] = await tx
       .select({ id: workspaceFiles.id })
       .from(workspaceFiles)
       .where(and(eq(workspaceFiles.id, fileId), workspaceFileScopeCondition(workspaceId, 'active')))
       .for('update')
       .limit(1)
     if (!file) throw new OrchestrationError('not_found', 'File not found')
-    const key = await deleteWorkspaceFileVersionInTx(tx, fileId, version)
-    return key ? enqueueWorkspaceFileStorageCleanups(tx, [key]) : null
+    const result = await deleteWorkspaceFileVersionInTx(tx, fileId, version)
+    return result.status === 'deleted'
+      ? {
+          status: result.status,
+          cleanupEventIds: await enqueueWorkspaceFileStorageCleanups(tx, [result.key]),
+        }
+      : { status: result.status, cleanupEventIds: [] }
   })
-  if (!cleanupEventIds) return false
 
-  await processWorkspaceFileStorageCleanupsNow(cleanupEventIds, {
+  await processWorkspaceFileStorageCleanupsNow(deletion.cleanupEventIds, {
     workspaceId,
     fileId,
     reason: 'deleted version',
   })
-  return true
+  return deletion.status
 }
 
 /**

apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox.ts

@@ -1,6 +1,6 @@
 import type { db } from '@sim/db'
 import { createLogger } from '@sim/logger'
-import { describeError, getErrorMessage } from '@sim/utils/errors'
+import { describeError } from '@sim/utils/errors'
 import { chunkArray } from '@sim/utils/helpers'
 import {
   enqueueOutboxEvents,
@@ -87,7 +87,7 @@ export async function processWorkspaceFileStorageCleanupsNow(
       logger.warn('Storage cleanup deferred after inline processing error', {
         ...logContext,
         eventId,
-        error: getErrorMessage(error),
+        error: describeError(error),
       })
     }
   }