fix skill memory bounding;

Author: icecrasher321

apps/sim/lib/workspaces/fork/copy/content-copy-runner.test.ts

@@ -2,7 +2,12 @@
  * @vitest-environment node
  */
 import { describe, expect, it } from 'vitest'
-import { serializeContentRefMaps } from '@/lib/workspaces/fork/copy/content-copy-runner'
+import {
+  hasForkContentToCopy,
+  serializeContentRefMaps,
+} from '@/lib/workspaces/fork/copy/content-copy-runner'
+import type { BlobCopyTask } from '@/lib/workspaces/fork/copy/copy-files'
+import type { ForkContentPlan } from '@/lib/workspaces/fork/copy/copy-resources'
 
 describe('serializeContentRefMaps', () => {
   it('converts each map to a record and drops empty maps', () => {
@@ -34,3 +39,46 @@ describe('serializeContentRefMaps', () => {
     expect(result.skills).toBeUndefined()
   })
 })
+
+describe('hasForkContentToCopy', () => {
+  const emptyPlan = (): ForkContentPlan => ({
+    sourceWorkspaceId: 'src',
+    childWorkspaceId: 'child',
+    userId: 'u',
+    tables: [],
+    knowledgeBases: [],
+    skills: [],
+    documents: [],
+  })
+  // The helper only inspects array lengths, so a single placeholder entry per kind is enough.
+  const oneSkill = [{}] as unknown as ForkContentPlan['skills']
+  const oneDoc = [{}] as unknown as ForkContentPlan['documents']
+  const oneTable = [{}] as unknown as ForkContentPlan['tables']
+  const oneKb = [{}] as unknown as ForkContentPlan['knowledgeBases']
+  const oneBlob = [{}] as unknown as BlobCopyTask[]
+  const noBlobs: BlobCopyTask[] = []
+
+  it('is true when skills are non-empty (the create-fork skill-only fix)', () => {
+    expect(hasForkContentToCopy({ ...emptyPlan(), skills: oneSkill }, noBlobs)).toBe(true)
+  })
+
+  it('is true when documents are non-empty', () => {
+    expect(hasForkContentToCopy({ ...emptyPlan(), documents: oneDoc }, noBlobs)).toBe(true)
+  })
+
+  it('is true when tables are non-empty', () => {
+    expect(hasForkContentToCopy({ ...emptyPlan(), tables: oneTable }, noBlobs)).toBe(true)
+  })
+
+  it('is true when knowledgeBases are non-empty', () => {
+    expect(hasForkContentToCopy({ ...emptyPlan(), knowledgeBases: oneKb }, noBlobs)).toBe(true)
+  })
+
+  it('is true when there are blob tasks', () => {
+    expect(hasForkContentToCopy(emptyPlan(), oneBlob)).toBe(true)
+  })
+
+  it('is false for an all-empty plan with no blob tasks', () => {
+    expect(hasForkContentToCopy(emptyPlan(), noBlobs)).toBe(false)
+  })
+})

apps/sim/lib/workspaces/fork/copy/content-copy-runner.ts

@@ -13,6 +13,26 @@ import type { ForkContentRefMaps } from '@/lib/workspaces/fork/remap/remap-conte
 
 const logger = createLogger('WorkspaceForkContentCopy')
 
+/**
+ * Whether a fork/sync has any heavy content to copy after the commit: table rows, KB documents,
+ * copied skill bodies, standalone documents, or file blobs. The single gate for scheduling the
+ * background content fill - shared by fork-create and promote so the two can't diverge. A
+ * skill-only (or documents-only) copy must still schedule it, otherwise the in-content `sim:` /
+ * serve-link rewrite in {@link runForkContentCopy} never runs and copied bodies keep source links.
+ */
+export function hasForkContentToCopy(
+  contentPlan: ForkContentPlan,
+  blobTasks: BlobCopyTask[]
+): boolean {
+  return (
+    contentPlan.tables.length > 0 ||
+    contentPlan.knowledgeBases.length > 0 ||
+    contentPlan.skills.length > 0 ||
+    contentPlan.documents.length > 0 ||
+    blobTasks.length > 0
+  )
+}
+
 /**
  * JSON-serializable form of {@link ForkContentRefMaps} (Maps become Records) so the in-content
  * reference maps survive the Trigger.dev payload boundary. Rehydrated to Maps in the runner.

apps/sim/lib/workspaces/fork/copy/copy-resources.test.ts

@@ -115,11 +115,14 @@ describe('copyForkResourceContent', () => {
     })
   })
 
-  it('#4 rewrites a copied skill body post-commit via db.update using the content maps', async () => {
+  it('#4 re-reads a copied skill body post-commit and rewrites it via db.update (never from payload)', async () => {
+    // The body is no longer carried in the plan - the content phase keyset-re-reads the child row.
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'child-skill-1', content: 'see [K](sim:knowledge/src-kb)' },
+    ])
+
     const result = await copyForkResourceContent({
-      contentPlan: basePlan({
-        skills: [{ childId: 'child-skill-1', content: 'see [K](sim:knowledge/src-kb)' }],
-      }),
+      contentPlan: basePlan({ skills: [{ childId: 'child-skill-1' }] }),
       contentRefMaps: { knowledgeBases: new Map([['src-kb', 'child-kb']]) },
       requestId: 'test',
     })
@@ -131,11 +134,13 @@ describe('copyForkResourceContent', () => {
     })
   })
 
-  it('#4 leaves a skill untouched when nothing in its body remaps', async () => {
+  it('#4 leaves a skill untouched when nothing in its re-read body remaps', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'child-skill-1', content: 'no references here' },
+    ])
+
     const result = await copyForkResourceContent({
-      contentPlan: basePlan({
-        skills: [{ childId: 'child-skill-1', content: 'no references here' }],
-      }),
+      contentPlan: basePlan({ skills: [{ childId: 'child-skill-1' }] }),
       contentRefMaps: { knowledgeBases: new Map([['src-kb', 'child-kb']]) },
       requestId: 'test',
     })
@@ -144,14 +149,14 @@ describe('copyForkResourceContent', () => {
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
   })
 
-  it('#4 skips the skill rewrite entirely when no content maps are supplied', async () => {
+  it('#4 skips the skill re-read + rewrite entirely when no content maps are supplied', async () => {
     await copyForkResourceContent({
-      contentPlan: basePlan({
-        skills: [{ childId: 'child-skill-1', content: 'see [K](sim:knowledge/src-kb)' }],
-      }),
+      contentPlan: basePlan({ skills: [{ childId: 'child-skill-1' }] }),
       requestId: 'test',
     })
 
+    // No maps -> the body is neither re-read from the DB nor updated.
+    expect(dbChainMockFns.select).not.toHaveBeenCalled()
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
   })
 
@@ -289,6 +294,67 @@ describe('copyForkResourceContainers custom-tool code env rewrite', () => {
   })
 })
 
+describe('copyForkResourceContainers skill copy', () => {
+  function makeSkillTx(rows: Array<Record<string, unknown>>) {
+    const inserted: Array<Record<string, unknown>> = []
+    const tx = {
+      select: () => ({ from: () => ({ where: () => Promise.resolve(rows) }) }),
+      insert: () => ({
+        values: (values: Array<Record<string, unknown>>) => {
+          inserted.push(...values)
+          return Promise.resolve()
+        },
+      }),
+    }
+    return { tx: tx as unknown as DbOrTx, inserted }
+  }
+
+  const skillSelection = {
+    customTools: [],
+    skills: ['sk-1'],
+    workflowMcpServers: [],
+    tables: [],
+    knowledgeBases: [],
+  }
+
+  it('copies the skill body IN-DB and carries only the child id in the content plan', async () => {
+    // The source projection deliberately omits `content` (it is copied server-side), so the row
+    // fed to the tx mock has none - the body must never be materialized in app memory here.
+    const { tx, inserted } = makeSkillTx([
+      {
+        id: 'sk-1',
+        name: 'My Skill',
+        description: 'desc',
+        workspaceId: 'src-ws',
+        userId: 'src-user',
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      },
+    ])
+
+    const result = await copyForkResourceContainers({
+      tx,
+      sourceWorkspaceId: 'src-ws',
+      childWorkspaceId: 'child-ws',
+      userId: 'user-1',
+      now: new Date(),
+      selection: skillSelection,
+      workflowIdMap: new Map(),
+    })
+
+    expect(inserted).toHaveLength(1)
+    const childId = inserted[0].id as string
+    expect(childId).not.toBe('sk-1')
+    expect(inserted[0].workspaceId).toBe('child-ws')
+    expect(inserted[0].userId).toBe('user-1')
+    // The body is deferred to a correlated subquery (in-DB copy), never a materialized string.
+    expect(typeof inserted[0].content).not.toBe('string')
+    // The content plan carries ONLY the child id - no skill body text crosses the job payload.
+    expect(result.contentPlan.skills).toEqual([{ childId }])
+    expect(result.names.skills).toEqual(['My Skill'])
+  })
+})
+
 describe('planForkMappedKbDocumentCopies', () => {
   const sourceRow = (id: string, knowledgeBaseId: string) => ({
     id,

apps/sim/lib/workspaces/fork/copy/copy-resources.ts

@@ -12,7 +12,7 @@ import {
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { and, asc, eq, gt, inArray, isNull, type SQL } from 'drizzle-orm'
+import { and, asc, eq, gt, inArray, isNull, type SQL, sql } from 'drizzle-orm'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import type { DbOrTx } from '@/lib/db/types'
 import type { TableSchema } from '@/lib/table/types'
@@ -47,6 +47,13 @@ const CONTENT_PAGE = 500
  */
 const KB_DOCUMENT_COPY_CONCURRENCY = 5
 
+/**
+ * Max copied skill bodies rewritten concurrently within one keyset page. Bounds the per-skill
+ * re-read + UPDATE fan-out so a page of copied skills doesn't issue every write at once; the keyset
+ * loop still processes one page at a time, so peak concurrency stays at this cap.
+ */
+const SKILL_REWRITE_CONCURRENCY = 5
+
 export interface CopyResourcesParams {
   tx: DbOrTx
   sourceWorkspaceId: string
@@ -95,13 +102,14 @@ export interface ForkContentKbEntry extends ForkContentPlanEntry {
 
 /**
  * A copied skill whose body's in-content references (`sim:` links + embedded file URLs) are
- * rewritten post-commit. The child row is inserted in the fork tx with the SOURCE body; the
- * content phase rewrites it best-effort so it points at the copied child resources (an unmapped
- * target degrades to a graceful broken link, never an FK/subblock reference).
+ * rewritten post-commit. Only the child id is carried: the child row is inserted in the fork tx
+ * with the SOURCE body copied IN-DB (never materialized in app memory or embedded in the job
+ * payload), and the content phase RE-READS the body keyset-paginated to rewrite it best-effort so
+ * it points at the copied child resources (an unmapped target degrades to a graceful broken link,
+ * never an FK/subblock reference).
  */
 export interface ForkContentSkillEntry {
   childId: string
-  content: string
 }
 
 /**
@@ -181,6 +189,13 @@ function setId(idMap: Map<ForkResourceType, Map<string, string>>, type: ForkReso
   return map
 }
 
+/**
+ * Child `skill` insert whose `content` is a correlated subquery (copied server-side from the source
+ * row) rather than a materialized string, so the fork tx never pulls skill bodies into app memory -
+ * see the skeleton skill copy in {@link copyForkResourceContainers}.
+ */
+type SkillSkeletonInsert = Omit<typeof skill.$inferInsert, 'content'> & { content: SQL }
+
 /**
  * Copy the selected resources' **container rows** into the child workspace inside
  * the fork transaction: custom tools, skills, and MCP server configs (each a
@@ -261,25 +276,37 @@ export async function copyForkResourceContainers(
   }
 
   if (selection.skills.length > 0) {
+    // Select every skill column EXCEPT `content`: the body (capped at 50 KB each, up to 2000 skills)
+    // is copied server-side via the correlated subquery below, so it is never materialized in app
+    // memory while the fork tx holds its locks - nor carried in the background-job payload.
     const rows = await tx
-      .select()
+      .select({
+        id: skill.id,
+        workspaceId: skill.workspaceId,
+        userId: skill.userId,
+        name: skill.name,
+        description: skill.description,
+        createdAt: skill.createdAt,
+        updatedAt: skill.updatedAt,
+      })
       .from(skill)
       .where(and(inArray(skill.id, selection.skills), eq(skill.workspaceId, sourceWorkspaceId)))
-    const inserts: (typeof skill.$inferInsert)[] = []
+    const inserts: SkillSkeletonInsert[] = []
     for (const row of rows) {
       const childId = generateId()
       inserts.push({
         ...row,
         id: childId,
         workspaceId: childWorkspaceId,
         userId,
+        // Copy the body straight from the source row in-DB (never through app memory). Re-read and
+        // rewritten post-commit (see copyForkResourceContent), out of the locked fork tx.
+        content: sql`(SELECT ${skill.content} FROM ${skill} WHERE ${skill.id} = ${row.id})`,
         createdAt: now,
         updatedAt: now,
       })
       record('skill', row.id, childId)
-      // Rewritten post-commit (see copyForkResourceContent): the row is inserted with the
-      // source body here so the locked fork tx never runs the per-skill content UPDATE.
-      contentPlan.skills.push({ childId, content: row.content })
+      contentPlan.skills.push({ childId })
       names.skills.push(row.name)
     }
     if (inserts.length > 0) await tx.insert(skill).values(inserts)
@@ -578,8 +605,9 @@ function remapTableRowResourceUrls(value: unknown, maps: ForkContentRefMaps): un
  * resource is copied in its own short statements (never one long transaction).
  * Best-effort: a failure on one resource is logged and the others continue - the
  * fork itself (workflows + container rows) already succeeded. Copied skill bodies are
- * rewritten here too (`contentRefMaps`), out of the locked fork tx; that rewrite is an
- * in-content link fixup, so a failure degrades to a broken link and never fails a resource.
+ * RE-READ (keyset-paginated) and rewritten here too (`contentRefMaps`), out of the locked
+ * fork tx; that rewrite is an in-content link fixup, so a failure degrades to a broken link
+ * and never fails a resource.
  */
 export async function copyForkResourceContent(params: {
   contentPlan: ForkContentPlan
@@ -760,26 +788,47 @@ export async function copyForkResourceContent(params: {
     }
   }
 
-  // Rewrite copied skill bodies out of the locked fork tx (the rows were inserted with the
-  // source body). Their `sim:` links + embedded file URLs are remapped to the child resources;
-  // this is an in-content link fixup, so a per-skill failure degrades to a broken link and is
-  // never counted as a failed resource (unmapped targets are left as graceful broken links).
+  // Rewrite copied skill bodies out of the locked fork tx (the rows were inserted with the source
+  // body via an in-DB copy). RE-READ each child body keyset-paginated - it is never carried in the
+  // job payload - remap its `sim:` links + embedded file URLs to the child resources, and write it
+  // back. An in-content link fixup: a per-skill failure degrades to a broken link and is never
+  // counted as a failed resource (unmapped targets are left as graceful broken links).
   if (contentRefMaps && contentPlan.skills.length > 0) {
-    for (const copiedSkill of contentPlan.skills) {
-      try {
-        const rewritten = rewriteForkContentRefs(copiedSkill.content, contentRefMaps)
-        if (rewritten !== copiedSkill.content) {
-          await db
-            .update(skill)
-            .set({ content: rewritten })
-            .where(eq(skill.id, copiedSkill.childId))
+    const childSkillIds = contentPlan.skills.map((entry) => entry.childId)
+    let afterId: string | null = null
+    for (;;) {
+      const where: SQL<unknown> | undefined =
+        afterId === null
+          ? inArray(skill.id, childSkillIds)
+          : and(inArray(skill.id, childSkillIds), gt(skill.id, afterId))
+      const rows = await db
+        .select({ id: skill.id, content: skill.content })
+        .from(skill)
+        .where(where)
+        .orderBy(asc(skill.id))
+        .limit(CONTENT_PAGE)
+      if (rows.length === 0) break
+      // Bounded fan-out: the mapper never rejects (it captures its own error), so mapWithConcurrency
+      // settles the whole page. A rewrite is a best-effort link fixup, so a per-skill failure is
+      // logged and the body keeps its source links rather than failing a resource.
+      await mapWithConcurrency(rows, SKILL_REWRITE_CONCURRENCY, async (row): Promise<void> => {
+        try {
+          const rewritten = rewriteForkContentRefs(row.content, contentRefMaps)
+          if (rewritten !== row.content) {
+            await db.update(skill).set({ content: rewritten }).where(eq(skill.id, row.id))
+          }
+        } catch (error) {
+          logger.warn(
+            `[${requestId}] Failed to rewrite copied skill content; keeping source links`,
+            {
+              childSkillId: row.id,
+              error: getErrorMessage(error),
+            }
+          )
         }
-      } catch (error) {
-        logger.warn(`[${requestId}] Failed to rewrite copied skill content; keeping source links`, {
-          childSkillId: copiedSkill.childId,
-          error: getErrorMessage(error),
-        })
-      }
+      })
+      afterId = rows[rows.length - 1].id
+      if (rows.length < CONTENT_PAGE) break
     }
   }