From ab9e2ae10056b3ccaa64945f53d6b9422edd9beb Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 18 Oct 2026 14:05:11 -0700 Subject: [PATCH 1/2] fix(custom-tools): skip title conflicts when persisting inline tools from a workflow save --- .../lib/workflows/custom-tools/operations.ts | 20 ++- .../custom-tools-persistence.integration.ts | 142 ++++++++++++++++++ .../persistence/custom-tools-persistence.ts | 7 + 3 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts diff --git a/apps/sim/lib/workflows/custom-tools/operations.ts b/apps/sim/lib/workflows/custom-tools/operations.ts index 7ebd763cbfd..bf897a61315 100644 --- a/apps/sim/lib/workflows/custom-tools/operations.ts +++ b/apps/sim/lib/workflows/custom-tools/operations.ts @@ -22,6 +22,13 @@ const logger = createLogger('CustomToolsOperations') export type CustomToolSortBy = 'title' | 'createdAt' | 'updatedAt' +/** + * What `upsertCustomTools` does with a tool it would insert under a title the + * workspace already holds: `'error'` throws and rolls back the whole batch, + * `'skip'` leaves the existing row untouched and persists the rest. + */ +type CustomToolTitleConflictPolicy = 'error' | 'skip' + /** * Internal function to create/update custom tools * Can be called from API routes or internal services @@ -36,8 +43,15 @@ export async function upsertCustomTools(params: { workspaceId: string userId: string requestId?: string + onTitleConflict?: CustomToolTitleConflictPolicy }) { - const { tools, workspaceId, userId, requestId = generateRequestId() } = params + const { + tools, + workspaceId, + userId, + requestId = generateRequestId(), + onTitleConflict = 'error', + } = params /** * Ahead of the transaction so a batch is rejected whole rather than landing @@ -106,6 +120,10 @@ export async function upsertCustomTools(params: { .limit(1) if (duplicateTitle.length > 0) { + if (onTitleConflict === 'skip') { + logger.info(`[${requestId}] Skipped custom tool "${tool.title}": title already exists`) + continue + } throw new Error(`A tool with the title "${tool.title}" already exists in this workspace`) } diff --git a/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts b/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts new file mode 100644 index 00000000000..091a3045235 --- /dev/null +++ b/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts @@ -0,0 +1,142 @@ +/** Workflow-save persistence of inline custom tools, against real PostgreSQL. */ +import { db } from '@sim/db' +import { customTools, user, workspace } from '@sim/db/schema' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { generateId } from '@sim/utils/id' +import { asc, eq } from 'drizzle-orm' +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' +import { upsertCustomTools } from '@/lib/workflows/custom-tools/operations' +import { extractAndPersistCustomTools } from '@/lib/workflows/persistence/custom-tools-persistence' + +const owner = `custom-tools-owner-${generateId()}` +const workspaceId = generateId() + +function declaration(name: string) { + return { + type: 'function' as const, + function: { + name, + description: `Runs ${name}`, + parameters: { type: 'object', properties: { id: { type: 'string' } }, required: ['id'] }, + }, + } +} + +/** An agent block carrying inline custom tools with no `toolId`, as an imported graph does. */ +function workflowStateWithInlineTools(tools: Array<{ title: string; code: string }>) { + return { + blocks: { + agent: { + type: 'agent', + subBlocks: { + tools: { + value: tools.map((tool) => ({ + type: 'custom-tool', + title: tool.title, + schema: declaration(tool.title.toLowerCase().replace(/\W+/g, '_')), + code: tool.code, + })), + }, + }, + }, + }, + } +} + +async function workspaceTools() { + return db + .select({ title: customTools.title, code: customTools.code }) + .from(customTools) + .where(eq(customTools.workspaceId, workspaceId)) + .orderBy(asc(customTools.title)) +} + +beforeAll(async () => { + const now = new Date() + await db.insert(user).values({ + id: owner, + name: 'Custom Tools', + email: `${owner}@custom-tools.test`, + emailVerified: true, + createdAt: now, + updatedAt: now, + }) + await insertWorkspaceFixture(db, [ + { id: workspaceId, name: 'Custom Tools', ownerId: owner, billedAccountUserId: owner }, + ]) +}) + +beforeEach(async () => { + await db.delete(customTools).where(eq(customTools.workspaceId, workspaceId)) +}) + +afterAll(async () => { + await db.delete(customTools).where(eq(customTools.workspaceId, workspaceId)) + await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) + await db.delete(user).where(eq(user.id, owner)) +}) + +describe('extractAndPersistCustomTools', () => { + it('re-saves a graph whose inline tool has no id and still persists the new tools beside it', async () => { + const first = await extractAndPersistCustomTools( + workflowStateWithInlineTools([{ title: 'Lookup order', code: 'return 1' }]), + workspaceId, + owner + ) + expect(first.errors).toEqual([]) + + const second = await extractAndPersistCustomTools( + workflowStateWithInlineTools([ + { title: 'Lookup order', code: 'return 1' }, + { title: 'Refund order', code: 'return 2' }, + ]), + workspaceId, + owner + ) + + expect(second.errors).toEqual([]) + expect(await workspaceTools()).toEqual([ + { title: 'Lookup order', code: 'return 1' }, + { title: 'Refund order', code: 'return 2' }, + ]) + }) + + it('leaves an existing workspace tool with the same title untouched', async () => { + await upsertCustomTools({ + tools: [{ title: 'Lookup order', schema: declaration('lookup'), code: 'return "library"' }], + workspaceId, + userId: owner, + }) + + const result = await extractAndPersistCustomTools( + workflowStateWithInlineTools([{ title: 'Lookup order', code: 'return "inline"' }]), + workspaceId, + owner + ) + + expect(result.errors).toEqual([]) + expect(await workspaceTools()).toEqual([{ title: 'Lookup order', code: 'return "library"' }]) + }) +}) + +describe('upsertCustomTools', () => { + it('rejects a create under a taken title and rolls back the rest of the batch', async () => { + await upsertCustomTools({ + tools: [{ title: 'Lookup order', schema: declaration('lookup'), code: 'return 1' }], + workspaceId, + userId: owner, + }) + + await expect( + upsertCustomTools({ + tools: [ + { title: 'Refund order', schema: declaration('refund'), code: 'return 2' }, + { title: 'Lookup order', schema: declaration('lookup'), code: 'return 3' }, + ], + workspaceId, + userId: owner, + }) + ).rejects.toThrow('A tool with the title "Lookup order" already exists in this workspace') + expect(await workspaceTools()).toEqual([{ title: 'Lookup order', code: 'return 1' }]) + }) +}) diff --git a/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts b/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts index 0fd2c9d7077..9e814c320ef 100644 --- a/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts +++ b/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts @@ -116,6 +116,12 @@ export function extractCustomToolsFromWorkflowState(workflowState: any): CustomT /** * Persist custom tools to the database using the upsert function * Creates new tools or updates existing ones + * + * @remarks + * An inline tool carries no `toolId` until the block is re-saved with one, so + * every save of the same graph re-offers it under a title its first save + * already inserted. A title conflict therefore skips that tool, leaving the + * existing row untouched, instead of rolling back the rest of the batch. */ export async function persistCustomToolsToDatabase( customToolsList: CustomTool[], @@ -167,6 +173,7 @@ export async function persistCustomToolsToDatabase( })), workspaceId, userId, + onTitleConflict: 'skip', }) saved = validTools.length From 727af4163a67ea7a705ff3ef38b0e3489f5789e6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 18 Oct 2026 14:14:06 -0700 Subject: [PATCH 2/2] fix(custom-tools): count skipped titles and skip title conflicts atomically --- .../lib/workflows/custom-tools/operations.ts | 226 +++++++++++------- .../custom-tools-persistence.integration.ts | 17 +- .../custom-tools-persistence.test.ts | 4 +- .../persistence/custom-tools-persistence.ts | 9 +- 4 files changed, 154 insertions(+), 102 deletions(-) diff --git a/apps/sim/lib/workflows/custom-tools/operations.ts b/apps/sim/lib/workflows/custom-tools/operations.ts index bf897a61315..23d8f5cf899 100644 --- a/apps/sim/lib/workflows/custom-tools/operations.ts +++ b/apps/sim/lib/workflows/custom-tools/operations.ts @@ -17,23 +17,13 @@ import { } from '@/lib/api/list-query' import { generateRequestId } from '@/lib/core/utils/request' import { assertValidCustomToolDeclaration } from '@/lib/custom-tools/schema' +import type { DbTransaction } from '@/lib/db/types' const logger = createLogger('CustomToolsOperations') export type CustomToolSortBy = 'title' | 'createdAt' | 'updatedAt' -/** - * What `upsertCustomTools` does with a tool it would insert under a title the - * workspace already holds: `'error'` throws and rolls back the whole batch, - * `'skip'` leaves the existing row untouched and persists the rest. - */ -type CustomToolTitleConflictPolicy = 'error' | 'skip' - -/** - * Internal function to create/update custom tools - * Can be called from API routes or internal services - */ -export async function upsertCustomTools(params: { +interface CustomToolWriteParams { tools: Array<{ id?: string title: string @@ -43,112 +33,162 @@ export async function upsertCustomTools(params: { workspaceId: string userId: string requestId?: string - onTitleConflict?: CustomToolTitleConflictPolicy -}) { - const { - tools, - workspaceId, - userId, - requestId = generateRequestId(), - onTitleConflict = 'error', - } = params - - /** - * Ahead of the transaction so a batch is rejected whole rather than landing - * the tools that preceded the unstorable one. - */ +} + +/** + * What a batch write does with a tool it would insert under a title the + * workspace already holds: `'error'` throws and rolls back the whole batch, + * `'skip'` leaves the existing row untouched and writes the rest. + */ +type CustomToolTitleConflictPolicy = 'error' | 'skip' + +/** + * Ahead of the transaction so a batch is rejected whole rather than landing + * the tools that preceded the unstorable one. + */ +function assertStorableBatch(tools: CustomToolWriteParams['tools']) { for (const tool of tools) { assertValidCustomToolDeclaration(tool.schema) } +} - return await db.transaction(async (tx) => { - for (const tool of tools) { - const nowTime = new Date() +/** Writes the batch inside `tx`, returning the titles skipped under the `'skip'` policy. */ +async function writeCustomTools( + tx: DbTransaction, + params: CustomToolWriteParams, + onTitleConflict: CustomToolTitleConflictPolicy +): Promise { + const { tools, workspaceId, userId, requestId = generateRequestId() } = params + const skippedTitles: string[] = [] + + for (const tool of tools) { + const nowTime = new Date() - if (tool.id) { - const existingWorkspaceTool = await tx - .select() - .from(customTools) + if (tool.id) { + const existingWorkspaceTool = await tx + .select() + .from(customTools) + .where(and(eq(customTools.id, tool.id), eq(customTools.workspaceId, workspaceId))) + .limit(1) + + if (existingWorkspaceTool.length > 0) { + await tx + .update(customTools) + .set({ + title: tool.title, + schema: tool.schema, + code: tool.code, + updatedAt: nowTime, + }) .where(and(eq(customTools.id, tool.id), eq(customTools.workspaceId, workspaceId))) - .limit(1) - - if (existingWorkspaceTool.length > 0) { - await tx - .update(customTools) - .set({ - title: tool.title, - schema: tool.schema, - code: tool.code, - updatedAt: nowTime, - }) - .where(and(eq(customTools.id, tool.id), eq(customTools.workspaceId, workspaceId))) - continue - } - - const existingLegacyTool = await tx - .select() - .from(customTools) - .where( - and( - eq(customTools.id, tool.id), - isNull(customTools.workspaceId), - eq(customTools.userId, userId) - ) - ) - .limit(1) - - if (existingLegacyTool.length > 0) { - await tx - .update(customTools) - .set({ - title: tool.title, - schema: tool.schema, - code: tool.code, - updatedAt: nowTime, - }) - .where(eq(customTools.id, tool.id)) - - logger.info(`[${requestId}] Updated legacy tool ${tool.id}`) - continue - } + continue } - const duplicateTitle = await tx + const existingLegacyTool = await tx .select() .from(customTools) - .where(and(eq(customTools.workspaceId, workspaceId), eq(customTools.title, tool.title))) + .where( + and( + eq(customTools.id, tool.id), + isNull(customTools.workspaceId), + eq(customTools.userId, userId) + ) + ) .limit(1) - if (duplicateTitle.length > 0) { - if (onTitleConflict === 'skip') { - logger.info(`[${requestId}] Skipped custom tool "${tool.title}": title already exists`) - continue - } - throw new Error(`A tool with the title "${tool.title}" already exists in this workspace`) + if (existingLegacyTool.length > 0) { + await tx + .update(customTools) + .set({ + title: tool.title, + schema: tool.schema, + code: tool.code, + updatedAt: nowTime, + }) + .where(eq(customTools.id, tool.id)) + + logger.info(`[${requestId}] Updated legacy tool ${tool.id}`) + continue } + } + + const row = { + id: generateShortId(), + workspaceId, + userId, + title: tool.title, + schema: tool.schema, + code: tool.code, + createdAt: nowTime, + updatedAt: nowTime, + } - await tx.insert(customTools).values({ - id: generateShortId(), - workspaceId, - userId, - title: tool.title, - schema: tool.schema, - code: tool.code, - createdAt: nowTime, - updatedAt: nowTime, - }) + if (onTitleConflict === 'skip') { + /** A conflict-target insert, so a title a concurrent save just committed is skipped too. */ + const inserted = await tx + .insert(customTools) + .values(row) + .onConflictDoNothing({ target: [customTools.workspaceId, customTools.title] }) + .returning({ id: customTools.id }) + if (inserted.length === 0) { + logger.info(`[${requestId}] Skipped custom tool "${tool.title}": title already exists`) + skippedTitles.push(tool.title) + } + continue } + const duplicateTitle = await tx + .select() + .from(customTools) + .where(and(eq(customTools.workspaceId, workspaceId), eq(customTools.title, tool.title))) + .limit(1) + + if (duplicateTitle.length > 0) { + throw new Error(`A tool with the title "${tool.title}" already exists in this workspace`) + } + + await tx.insert(customTools).values(row) + } + + return skippedTitles +} + +/** + * Internal function to create/update custom tools + * Can be called from API routes or internal services + */ +export async function upsertCustomTools(params: CustomToolWriteParams) { + assertStorableBatch(params.tools) + + return await db.transaction(async (tx) => { + await writeCustomTools(tx, params, 'error') + const resultTools = await tx .select() .from(customTools) - .where(eq(customTools.workspaceId, workspaceId)) + .where(eq(customTools.workspaceId, params.workspaceId)) .orderBy(desc(customTools.createdAt)) return resultTools }) } +/** + * `upsertCustomTools` for writes that re-offer tools they may already have + * stored (workflow saves and imports): a tool whose title the workspace already + * holds is skipped, leaving that row untouched, instead of failing the batch. + * + * @returns The titles that were skipped. + */ +export async function upsertCustomToolsSkippingTakenTitles( + params: CustomToolWriteParams +): Promise<{ skippedTitles: string[] }> { + assertStorableBatch(params.tools) + + const skippedTitles = await db.transaction((tx) => writeCustomTools(tx, params, 'skip')) + return { skippedTitles } +} + export async function listCustomTools(params: { userId: string; workspaceId?: string }) { const { userId, workspaceId } = params return workspaceId diff --git a/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts b/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts index 091a3045235..901a9f7255c 100644 --- a/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts +++ b/apps/sim/lib/workflows/persistence/custom-tools-persistence.integration.ts @@ -94,7 +94,7 @@ describe('extractAndPersistCustomTools', () => { owner ) - expect(second.errors).toEqual([]) + expect(second).toEqual({ saved: 1, errors: [] }) expect(await workspaceTools()).toEqual([ { title: 'Lookup order', code: 'return 1' }, { title: 'Refund order', code: 'return 2' }, @@ -114,9 +114,22 @@ describe('extractAndPersistCustomTools', () => { owner ) - expect(result.errors).toEqual([]) + expect(result).toEqual({ saved: 0, errors: [] }) expect(await workspaceTools()).toEqual([{ title: 'Lookup order', code: 'return "library"' }]) }) + + it('lets concurrent saves of the same new inline tool both succeed with one row', async () => { + const state = workflowStateWithInlineTools([{ title: 'Lookup order', code: 'return 1' }]) + + const results = await Promise.all([ + extractAndPersistCustomTools(state, workspaceId, owner), + extractAndPersistCustomTools(state, workspaceId, owner), + ]) + + expect(results.map((result) => result.errors)).toEqual([[], []]) + expect(results.map((result) => result.saved).sort()).toEqual([0, 1]) + expect(await workspaceTools()).toEqual([{ title: 'Lookup order', code: 'return 1' }]) + }) }) describe('upsertCustomTools', () => { diff --git a/apps/sim/lib/workflows/persistence/custom-tools-persistence.test.ts b/apps/sim/lib/workflows/persistence/custom-tools-persistence.test.ts index 24b88216159..2709962af4f 100644 --- a/apps/sim/lib/workflows/persistence/custom-tools-persistence.test.ts +++ b/apps/sim/lib/workflows/persistence/custom-tools-persistence.test.ts @@ -5,7 +5,7 @@ const { mockUpsertCustomTools } = vi.hoisted(() => ({ })) vi.mock('@/lib/workflows/custom-tools/operations', () => ({ - upsertCustomTools: mockUpsertCustomTools, + upsertCustomToolsSkippingTakenTitles: mockUpsertCustomTools, })) import { persistCustomToolsToDatabase } from '@/lib/workflows/persistence/custom-tools-persistence' @@ -34,7 +34,7 @@ function customTool(overrides: Record = {}) { describe('persistCustomToolsToDatabase', () => { beforeEach(() => { - mockUpsertCustomTools.mockResolvedValue([]) + mockUpsertCustomTools.mockResolvedValue({ skippedTitles: [] }) }) it('skips a declaration missing the function discriminator the public API republishes', async () => { diff --git a/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts b/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts index 9e814c320ef..728c14f1f0f 100644 --- a/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts +++ b/apps/sim/lib/workflows/persistence/custom-tools-persistence.ts @@ -1,7 +1,7 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { isValidCustomToolDeclaration } from '@/lib/custom-tools/schema' -import { upsertCustomTools } from '@/lib/workflows/custom-tools/operations' +import { upsertCustomToolsSkippingTakenTitles } from '@/lib/workflows/custom-tools/operations' const logger = createLogger('CustomToolsPersistence') @@ -164,7 +164,7 @@ export async function persistCustomToolsToDatabase( } try { - await upsertCustomTools({ + const { skippedTitles } = await upsertCustomToolsSkippingTakenTitles({ tools: validTools.map((tool) => ({ id: tool.toolId, title: tool.title, @@ -173,11 +173,10 @@ export async function persistCustomToolsToDatabase( })), workspaceId, userId, - onTitleConflict: 'skip', }) - saved = validTools.length - logger.info(`Persisted ${saved} custom tool(s)`, { workspaceId }) + saved = validTools.length - skippedTitles.length + logger.info(`Persisted ${saved} custom tool(s)`, { workspaceId, skippedTitles }) } catch (error) { const errorMsg = `Failed to persist custom tools: ${toError(error).message}` logger.error(errorMsg, { error })