From 19596442cf1193e7008c35376258d20ffc51c6ef Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 3 Nov 2026 12:55:01 -0800 Subject: [PATCH 1/3] fix(api): classify transient database reads and client-caused failures instead of answering 500 --- .../app/api/files/serve/[...path]/route.ts | 15 +++- apps/sim/app/api/logs/export/route.ts | 5 +- .../mothership/integrations/catalog/route.ts | 12 ++- .../workspaces/[id]/background-work/route.ts | 9 +- .../app/api/workspaces/members/[id]/route.ts | 3 +- apps/sim/lib/core/utils/stream-limits.test.ts | 26 ++++++ apps/sim/lib/core/utils/stream-limits.ts | 21 ++++- .../lib/core/utils/with-route-handler.test.ts | 86 +++++++++++++++++++ apps/sim/lib/core/utils/with-route-handler.ts | 45 +++++++++- apps/sim/lib/mcp/utils.ts | 28 ++++-- .../sim/lib/mothership/chat/lifecycle.test.ts | 11 ++- apps/sim/lib/mothership/chat/lifecycle.ts | 7 +- .../contexts/workspace/file-folder-manager.ts | 8 ++ .../apply-workflow-operations.test.ts | 10 +++ .../application/apply-workflow-operations.ts | 25 ++---- apps/sim/lib/workflows/editing/engine.ts | 9 +- .../lib/workflows/editing/operations.test.ts | 34 ++++++++ apps/sim/lib/workflows/editing/types.ts | 6 ++ 18 files changed, 319 insertions(+), 41 deletions(-) diff --git a/apps/sim/app/api/files/serve/[...path]/route.ts b/apps/sim/app/api/files/serve/[...path]/route.ts index df007d91f25..f29b44ffa65 100644 --- a/apps/sim/app/api/files/serve/[...path]/route.ts +++ b/apps/sim/app/api/files/serve/[...path]/route.ts @@ -19,6 +19,7 @@ import type { StorageContext } from '@/lib/uploads/config' import { readOrganizationChatAttachment } from '@/lib/uploads/contexts/organization-assistant/application' import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace' import { parseWorkspaceFileKey } from '@/lib/uploads/contexts/workspace/file-manager' +import { hasObjectNotFoundCause } from '@/lib/uploads/core/errors' import { downloadFile, downloadFileStream, headObject } from '@/lib/uploads/core/storage-service' import { resolveServableDocBytes } from '@/lib/uploads/documents/compile' import { DocCompileUserError } from '@/lib/uploads/documents/compile-error' @@ -34,6 +35,7 @@ import { inferContextFromKey, isRenderableDocumentName, resolveEffectiveMimeType, + tryInferContextFromKey, } from '@/lib/uploads/utils/file' import { internalWorkspaceFileServeAuth } from '@/lib/workspace-files/api' import { fetchAuthorizedServableWorkspaceFileBuffer } from '@/lib/workspace-files/application/fetch-servable-workspace-file-buffer' @@ -71,7 +73,11 @@ const logger = createLogger('FilesServeAPI') * its way out of, not something on call needs to look at. */ function logServeFailure(message: string, error: unknown): void { - if (error instanceof FileNotFoundError || isPayloadSizeLimitError(error)) { + if ( + error instanceof FileNotFoundError || + hasObjectNotFoundCause(error) || + isPayloadSizeLimitError(error) + ) { logger.info(message, { reason: getErrorMessage(error) }) return } @@ -389,6 +395,9 @@ export const GET = withRouteHandler( }) } + /** No storage context owns a key without a known prefix, so no file can live there. */ + if (!tryInferContextFromKey(cloudKey)) throw new FileNotFoundError('File not found') + const isPublicByKeyPrefix = cloudKey.startsWith('profile-pictures/') || cloudKey.startsWith('og-images/') || @@ -536,6 +545,10 @@ export const GET = withRouteHandler( return createFileErrorResponse(error) } + if (hasObjectNotFoundCause(error)) { + return createFileErrorResponse(new FileNotFoundError('File not found')) + } + return createFileErrorResponse( error instanceof Error ? error : new Error('Failed to serve file') ) diff --git a/apps/sim/app/api/logs/export/route.ts b/apps/sim/app/api/logs/export/route.ts index 73468428ae4..9ab1521ad82 100644 --- a/apps/sim/app/api/logs/export/route.ts +++ b/apps/sim/app/api/logs/export/route.ts @@ -4,6 +4,7 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { and, desc, eq, lt, or, type SQL, sql } from 'drizzle-orm' import { type NextRequest, NextResponse } from 'next/server' +import { validationErrorResponse } from '@/lib/api/server/validation' import { getSession } from '@/lib/auth' import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency' import { formatCsvValue, toCsvLinePieces, toCsvRow } from '@/lib/core/utils/csv' @@ -50,7 +51,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const userId = session.user.id const { searchParams } = new URL(request.url) - const params = LogFilterParamsSchema.parse(Object.fromEntries(searchParams.entries())) + const parsedParams = LogFilterParamsSchema.safeParse(Object.fromEntries(searchParams.entries())) + if (!parsedParams.success) return validationErrorResponse(parsedParams.error) + const params = parsedParams.data const selectColumns = { id: workflowExecutionLogs.id, diff --git a/apps/sim/app/api/mothership/integrations/catalog/route.ts b/apps/sim/app/api/mothership/integrations/catalog/route.ts index 0db6e5d49c1..570eebbed71 100644 --- a/apps/sim/app/api/mothership/integrations/catalog/route.ts +++ b/apps/sim/app/api/mothership/integrations/catalog/route.ts @@ -1,9 +1,11 @@ import { readIntegrationCatalogContract } from '@/lib/api/contracts/mothership-integrations' import { defineInternalJsonRoute, + internalErrorResponse, internalOrchestrationErrorPolicy, internalRateLimits, } from '@/lib/api/server/routes' +import { categorizeMcpServerError } from '@/lib/mcp/utils' import { internalCopilotAuth } from '@/lib/mothership/auth/internal' import { INTEGRATION_CATALOG_AUDIENCE, @@ -18,7 +20,15 @@ export const POST = defineInternalJsonRoute({ rateLimit: internalRateLimits.none({ reason: 'Private bounded catalog discovery rechecks current authorization on every request.', }), - errorPolicy: internalOrchestrationErrorPolicy, + errorPolicy: { + /** A targeted MCP server that cannot be reached is the server's failure, not Sim's. */ + project(error) { + const mcpFailure = categorizeMcpServerError(error) + if (mcpFailure) return internalErrorResponse(mcpFailure.status, { error: mcpFailure.message }) + return internalOrchestrationErrorPolicy.project(error) + }, + unhandled: internalOrchestrationErrorPolicy.unhandled, + }, mapInput: ({ body }) => body, useCase: readIntegrationCatalog, }) diff --git a/apps/sim/app/api/workspaces/[id]/background-work/route.ts b/apps/sim/app/api/workspaces/[id]/background-work/route.ts index 727b4445f78..550d5a09fca 100644 --- a/apps/sim/app/api/workspaces/[id]/background-work/route.ts +++ b/apps/sim/app/api/workspaces/[id]/background-work/route.ts @@ -2,6 +2,7 @@ import { db } from '@sim/db' import { type NextRequest, NextResponse } from 'next/server' import { getWorkspaceBackgroundWorkContract } from '@/lib/api/contracts/workspace-fork' import { parseRequest } from '@/lib/api/server' +import { orchestrationFailureResponse } from '@/lib/api/server/orchestration-response' import { getSession } from '@/lib/auth' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { listSurfacedBackgroundWork } from '@/ee/workspace-forking/lib/background-work/store' @@ -21,7 +22,13 @@ export const GET = withRouteHandler( // The fork Activity feed is a fork feature: gate it behind the same forking-enabled + // workspace-admin check the other fork routes use, instead of a bare access check. - await assertWorkspaceAdminAccess(id, session.user.id) + try { + await assertWorkspaceAdminAccess(id, session.user.id) + } catch (error) { + const refused = orchestrationFailureResponse(error) + if (refused) return refused + throw error + } const { rows, nextCursor } = await listSurfacedBackgroundWork(db, id, { cursor, limit }) return NextResponse.json({ diff --git a/apps/sim/app/api/workspaces/members/[id]/route.ts b/apps/sim/app/api/workspaces/members/[id]/route.ts index 19e84fcd0c1..0c9a109c23c 100644 --- a/apps/sim/app/api/workspaces/members/[id]/route.ts +++ b/apps/sim/app/api/workspaces/members/[id]/route.ts @@ -2,6 +2,7 @@ import { removeWorkspaceMemberContract } from '@/lib/api/contracts/invitations' import { defineInternalJsonRoute, internalErrorResponse, + internalOrchestrationErrorPolicy, internalRateLimits, internalSessionAuth, } from '@/lib/api/server/routes' @@ -32,7 +33,7 @@ export const DELETE = defineInternalJsonRoute({ error instanceof NoWorkspaceAccessError ) return internalErrorResponse(403, { error: 'Insufficient permissions' }) - return null + return internalOrchestrationErrorPolicy.project(error) }, unhandled: () => internalErrorResponse(500, { error: 'Failed to remove workspace member' }), }, diff --git a/apps/sim/lib/core/utils/stream-limits.test.ts b/apps/sim/lib/core/utils/stream-limits.test.ts index 156b821da52..e7eb79beace 100644 --- a/apps/sim/lib/core/utils/stream-limits.test.ts +++ b/apps/sim/lib/core/utils/stream-limits.test.ts @@ -293,6 +293,32 @@ describe('stream limits', () => { } ) + it.each([ + ['a JSON body', 'application/json', '{"name":"example"}', true], + [ + 'a boundary the body never uses', + 'multipart/form-data; boundary=declared', + '--other\r\n', + true, + ], + ['a streamed body the parser cannot decode', 'text/plain', 'name=example', false], + ])('rejects %s as invalid multipart input', async (_label, contentType, body, declareLength) => { + const bytes = new TextEncoder().encode(body) + const request = new Request('http://localhost/upload', { + method: 'POST', + headers: { + 'content-type': contentType, + ...(declareLength ? { 'content-length': String(bytes.byteLength) } : {}), + }, + body: declareLength ? bytes : streamFromChunks([bytes]), + duplex: 'half', + } as RequestInit) + + await expect( + readFormDataWithLimit(request, { maxBytes: 1024 * 1024, label: 'multipart body' }) + ).rejects.toBeInstanceOf(MultipartFieldValidationError) + }) + it('rejects multipart streams without content-length once bytes exceed the limit', async () => { const request = new Request('http://localhost/upload', { method: 'POST', diff --git a/apps/sim/lib/core/utils/stream-limits.ts b/apps/sim/lib/core/utils/stream-limits.ts index 18af9773f79..7587782f69b 100644 --- a/apps/sim/lib/core/utils/stream-limits.ts +++ b/apps/sim/lib/core/utils/stream-limits.ts @@ -138,7 +138,7 @@ export async function readFormDataWithLimit( assertContentLengthWithinLimit(request.headers, options.maxBytes, options.label) if (request.headers?.get('content-length') || !request.body) { - const formData = await request.formData() + const formData = await parseFormData(request) assertMultipartFieldsAreStorable(formData) return formData } @@ -149,11 +149,28 @@ export async function readFormDataWithLimit( headers: request.headers, body: new Uint8Array(body), }) - const formData = await boundedRequest.formData() + const formData = await parseFormData(boundedRequest) assertMultipartFieldsAreStorable(formData) return formData } +/** + * The fetch body parser rejects a body that does not decode as the content type it + * declares (a JSON body, a boundary the body never uses) with a bare `TypeError`. That + * is the caller's malformed input, so it surfaces as the same validation failure every + * multipart route already answers with a 400. + */ +async function parseFormData(request: Pick): Promise { + try { + return await request.formData() + } catch (error) { + if (error instanceof TypeError) { + throw new MultipartFieldValidationError('Request body must be valid multipart form data') + } + throw error + } +} + export interface ReadStreamWithLimitOptions { maxBytes: number label: string diff --git a/apps/sim/lib/core/utils/with-route-handler.test.ts b/apps/sim/lib/core/utils/with-route-handler.test.ts index 40fccd3d4e6..d1dae49dae3 100644 --- a/apps/sim/lib/core/utils/with-route-handler.test.ts +++ b/apps/sim/lib/core/utils/with-route-handler.test.ts @@ -1,4 +1,5 @@ import { loggerMock } from '@sim/testing' +import { DrizzleQueryError } from 'drizzle-orm/errors' import { NextRequest, NextResponse } from 'next/server' import { describe, expect, it, vi } from 'vitest' import { HttpError } from '@/lib/core/utils/http-error' @@ -142,4 +143,89 @@ describe('withRouteHandler', () => { await expect(response.json()).resolves.toEqual({ family: 'generic' }) } ) + + describe('a database that is temporarily unavailable', () => { + const SQL = 'select "secret_column" from "workflow_execution_logs" where "id" = $1' + + function failedQuery(code: string, message: string): DrizzleQueryError { + return new DrizzleQueryError(SQL, ['id-1'], Object.assign(new Error(message), { code })) + } + + function familyRoute(error: unknown) { + return withRouteHandler( + async () => { + throw error + }, + { + typedErrorResponse: ({ error: typed, status }) => + NextResponse.json({ family: 'typed', error: typed.message }, { status }), + unhandledErrorResponse: () => NextResponse.json({ family: 'generic' }, { status: 500 }), + } + ) + } + + it.each([ + [ + 'a replica recovery conflict', + failedQuery('40001', 'canceling statement due to conflict with recovery'), + ], + ['a statement timeout', failedQuery('57014', 'canceling statement due to statement timeout')], + [ + 'a connection lost under the query', + failedQuery('CONNECTION_CLOSED', 'write CONNECTION_CLOSED'), + ], + [ + 'a server restarting', + failedQuery('57P01', 'terminating connection due to administrator command'), + ], + ])('answers a read failing on %s with a retryable 503', async (_, error) => { + const response = await familyRoute(error)( + new NextRequest('http://localhost/api/test'), + undefined + ) + + expect(response.status).toBe(503) + expect(Number(response.headers.get('Retry-After'))).toBeGreaterThan(0) + const body = await response.json() + expect(body.family).toBe('typed') + expect(JSON.stringify(body)).not.toContain('secret_column') + }) + + it('answers a HEAD the same way as its GET', async () => { + const response = await familyRoute( + failedQuery('40001', 'canceling statement due to conflict with recovery') + )(new NextRequest('http://localhost/api/test', { method: 'HEAD' }), undefined) + + expect(response.status).toBe(503) + }) + + it.each(['POST', 'PUT', 'PATCH', 'DELETE'])( + 'leaves a %s unhandled, since an earlier statement may already have committed', + async (method) => { + const response = await familyRoute( + failedQuery('40001', 'canceling statement due to conflict with recovery') + )(new NextRequest('http://localhost/api/test', { method }), undefined) + + expect(response.status).toBe(500) + expect(response.headers.get('Retry-After')).toBeNull() + } + ) + + it.each([ + ['a user-requested cancel', failedQuery('57014', 'canceling statement due to user request')], + ['a permanent query error', failedQuery('42601', 'syntax error at or near ")"')], + [ + 'a socket reset outside any query', + Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }), + ], + ])('leaves %s on a read unhandled', async (_, error) => { + const response = await familyRoute(error)( + new NextRequest('http://localhost/api/test'), + undefined + ) + + expect(response.status).toBe(500) + await expect(response.json()).resolves.toEqual({ family: 'generic' }) + }) + }) }) diff --git a/apps/sim/lib/core/utils/with-route-handler.ts b/apps/sim/lib/core/utils/with-route-handler.ts index d0d2110322e..7a5fc3889c4 100644 --- a/apps/sim/lib/core/utils/with-route-handler.ts +++ b/apps/sim/lib/core/utils/with-route-handler.ts @@ -1,10 +1,17 @@ import { createLogger, runWithRequestContext } from '@sim/logger' import { resolveClientInfo } from '@sim/utils/client-info' -import { describeError, findCause, getErrorMessage, redactBoundParameters } from '@sim/utils/errors' +import { + describeError, + findCause, + getErrorMessage, + getTransientDatabaseFailure, + redactBoundParameters, +} from '@sim/utils/errors' import type { NextRequest } from 'next/server' import { NextResponse } from 'next/server' import { hasExternalApiCredentials } from '@/lib/api/server/credential-headers' import { getRateLimitHeaders } from '@/lib/api/server/rate-limit-context' +import { ADMISSION_RETRY_AFTER_SECONDS } from '@/lib/core/admission/transient-failure' import { FILE_DELIVERY_CSP_PATH_PATTERN, getMainCSPPolicy } from '@/lib/core/security/csp' import { HttpError } from '@/lib/core/utils/http-error' import { generateRequestId } from '@/lib/core/utils/request' @@ -61,6 +68,34 @@ function readTypedError(error: unknown): RouteHandlerTypedErrorContext['error'] return error } +class DatabaseTemporarilyUnavailableError extends HttpError { + readonly statusCode = 503 + + constructor() { + super('Service temporarily unavailable. Please try again.') + this.name = 'DatabaseTemporarilyUnavailableError' + } +} + +/** Methods with no side effects (RFC 9110 ยง9.2.1), so a failed attempt left nothing behind. */ +const SAFE_METHODS = new Set(['GET', 'HEAD']) + +/** + * A safe request whose query failed for a reason that clears on its own โ€” a replica + * recovery conflict, a statement or lock timeout, a lost connection, a server restart โ€” + * answers a retryable 503 rather than an opaque 500. Writes stay unhandled: an earlier + * statement in the same request may already have committed, so "retry later" would be + * unsafe advice. The classifier only counts socket errors that a database query carried, + * so an upstream HTTP reset is never reported as a database outage. + */ +function readDatabaseUnavailableError( + method: string, + error: unknown +): DatabaseTemporarilyUnavailableError | undefined { + if (!SAFE_METHODS.has(method) || !getTransientDatabaseFailure(error)) return undefined + return new DatabaseTemporarilyUnavailableError() +} + /** * Stamps the request id, plus the rate-limit trio when the route consulted a * bucket for this request. Applied on both the success and the unhandled-error @@ -198,7 +233,8 @@ export function withRouteHandler( return response } - const typedError = readTypedError(error) + const databaseUnavailable = readDatabaseUnavailableError(method, error) + const typedError = readTypedError(error) ?? databaseUnavailable if (typedError) { const typedStatus = typedError.statusCode if (typedStatus >= 500) { @@ -213,7 +249,10 @@ export function withRouteHandler( } response = options.typedErrorResponse ? options.typedErrorResponse({ error: typedError, requestId, status: typedStatus }) - : NextResponse.json({ error: message, requestId }, { status: typedStatus }) + : NextResponse.json({ error: typedError.message, requestId }, { status: typedStatus }) + if (databaseUnavailable && !response.headers.has('Retry-After')) { + response.headers.set('Retry-After', ADMISSION_RETRY_AFTER_SECONDS.toString()) + } applyResponseHeaders(response, request, requestId) return response } diff --git a/apps/sim/lib/mcp/utils.ts b/apps/sim/lib/mcp/utils.ts index b8ab227c82b..a7dcbab1ffd 100644 --- a/apps/sim/lib/mcp/utils.ts +++ b/apps/sim/lib/mcp/utils.ts @@ -163,15 +163,13 @@ export function isMcpServerUnavailableError(error: unknown): boolean { } /** - * Enhanced error categorization for more specific HTTP status codes. - * Returns safe, generic messages to prevent leaking internal details. + * Status and safe message for a failure raised by one of Sim's own MCP server error + * classes, or `null` for anything else. Unlike {@link categorizeError} it never guesses + * from message text, so a caller can apply it where an unrelated error must stay a 500. */ -export function categorizeError(error: unknown): { message: string; status: number } { - if (!(error instanceof Error)) { - return { message: 'Unknown error occurred', status: 500 } - } - - // Typed dispatch first โ€” our own classes carry definitive intent. +export function categorizeMcpServerError( + error: unknown +): { message: string; status: number } | null { if (error instanceof McpOauthAuthorizationRequiredError || error instanceof UnauthorizedError) { return { message: 'Authentication required', status: 401 } } @@ -187,6 +185,20 @@ export function categorizeError(error: unknown): { message: string; status: numb if (error instanceof McpSsrfError || error instanceof McpDomainNotAllowedError) { return { message: 'Server URL is not allowed by egress policy', status: 403 } } + return null +} + +/** + * Enhanced error categorization for more specific HTTP status codes. + * Returns safe, generic messages to prevent leaking internal details. + */ +export function categorizeError(error: unknown): { message: string; status: number } { + if (!(error instanceof Error)) { + return { message: 'Unknown error occurred', status: 500 } + } + + const typed = categorizeMcpServerError(error) + if (typed) return typed // Fall back to substring matching for SDK / third-party errors we don't // own a typed class for. diff --git a/apps/sim/lib/mothership/chat/lifecycle.test.ts b/apps/sim/lib/mothership/chat/lifecycle.test.ts index f02fef1fe8b..3a2aaae4b30 100644 --- a/apps/sim/lib/mothership/chat/lifecycle.test.ts +++ b/apps/sim/lib/mothership/chat/lifecycle.test.ts @@ -34,7 +34,7 @@ import { resolveOrCreateChat, } from '@/lib/mothership/chat/lifecycle' -const CHAT_ID = 'chat-1' +const CHAT_ID = 'c7a4d2e1-5b3f-4a8e-9c6d-1f2e3a4b5c6d' const USER_ID = 'user-1' // A chat with no workflow/workspace skips the authz lookups and authorizes directly. @@ -236,6 +236,15 @@ describe('organization chat isolation', () => { }) }) + it.each(['c7a4d2e1-5b3f-4a8e-9c6d-1f2e3a4b5c6d.', 'chat-1'])( + 'answers a malformed chat id %s as not found without querying', + async (chatId) => { + expect(await getAccessibleCopilotChatWithMessages(chatId, USER_ID)).toBeNull() + expect(await getAccessibleCopilotChatAuth(chatId, USER_ID)).toBeNull() + expect(dbChainMockFns.limit).not.toHaveBeenCalled() + } + ) + it('does not expose org content to a legacy caller without a principal', async () => { dbChainMockFns.limit.mockResolvedValueOnce([{ ...chatRow, organizationId: 'org-1' }]) expect(await getAccessibleCopilotChatWithMessages(CHAT_ID, USER_ID)).toBeNull() diff --git a/apps/sim/lib/mothership/chat/lifecycle.ts b/apps/sim/lib/mothership/chat/lifecycle.ts index 915639411dc..feb8ac384d6 100644 --- a/apps/sim/lib/mothership/chat/lifecycle.ts +++ b/apps/sim/lib/mothership/chat/lifecycle.ts @@ -6,6 +6,7 @@ import { authorizeWorkflowByWorkspacePermission, getActiveWorkflowRecord, } from '@sim/platform-authz/workflow' +import { isValidUuid } from '@sim/utils/id' import { and, asc, desc, eq, isNull, type SQL, sql } from 'drizzle-orm' import { preciseTimestampText } from '@/lib/api/list-query' import { asOrchestrationError } from '@/lib/core/orchestration/types' @@ -291,7 +292,9 @@ export async function loadChatMcpServerIds(chatId: string): Promise { /** * Ownership + liveness predicate shared by the accessible-chat loaders: - * the chat must belong to the user and not be soft-deleted. + * the chat must belong to the user and not be soft-deleted. Loaders check the id + * is a UUID first: a malformed one (a URL pasted with trailing punctuation) can + * match no chat, and Postgres would reject it with 22P02 instead of no row. */ function ownedLiveChatWhere(chatId: string, userId: string) { return and( @@ -451,6 +454,7 @@ async function loadAccessibleCopilotChatAuth( | typeof authorizeOrganizationChat | typeof authorizeOrganizationChatCancellation ): Promise { + if (!isValidUuid(chatId)) return null const [chat] = await db .select(copilotChatAuthColumns) .from(copilotChats) @@ -471,6 +475,7 @@ async function getAccessibleCopilotChatDetail( userId: string, principal?: Principal ): Promise { + if (!isValidUuid(chatId)) return null const [chat] = await db .select(copilotChatDetailColumns) .from(copilotChats) diff --git a/apps/sim/lib/uploads/contexts/workspace/file-folder-manager.ts b/apps/sim/lib/uploads/contexts/workspace/file-folder-manager.ts index cb42467105f..a43c1772144 100644 --- a/apps/sim/lib/uploads/contexts/workspace/file-folder-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/file-folder-manager.ts @@ -1177,7 +1177,9 @@ export async function moveFileItems( throw new WorkspaceFileItemsNotFoundError(missingFileIds, missingFolderIds) } + const movingFileNameCounts = new Map() for (const file of movingFiles) { + movingFileNameCounts.set(file.name, (movingFileNameCounts.get(file.name) ?? 0) + 1) const conflictingFiles = await tx .select({ id: workspaceFiles.id }) .from(workspaceFiles) @@ -1196,6 +1198,12 @@ export async function moveFileItems( } } + for (const [name, count] of movingFileNameCounts) { + if (count > 1) { + throw new WorkspaceFileMoveConflictError(name) + } + } + const movingFolderNameCounts = new Map() for (const folder of movingFolders) { movingFolderNameCounts.set(folder.name, (movingFolderNameCounts.get(folder.name) ?? 0) + 1) diff --git a/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts b/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts index 490cf941a9a..b5121017a62 100644 --- a/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts +++ b/apps/sim/lib/workflows/application/apply-workflow-operations.test.ts @@ -214,6 +214,7 @@ describe('applyWorkflowOperations', () => { state: graph(), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, mintedBlockIds: {}, }) mocks.collectReferences.mockResolvedValue([]) @@ -239,6 +240,7 @@ describe('applyWorkflowOperations', () => { state: graph({ 'block-1': BLOCK }), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, mintedBlockIds: {}, }) mocks.lintGraph.mockReturnValue({ ...EMPTY_GRAPH_LINT, orphanBlocks: [orphan] }) @@ -272,6 +274,7 @@ describe('applyWorkflowOperations', () => { state: graph(), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, mintedBlockIds: { triage: 'preview-uuid' }, }) @@ -361,6 +364,7 @@ describe('applyWorkflowOperations', () => { reason: 'Name taken', }, ], + declinedOperationCount: 1, }) const result = await applyWorkflowOperations.execute({ @@ -385,6 +389,7 @@ describe('applyWorkflowOperations', () => { reason: 'Target not created yet', }, ], + declinedOperationCount: 0, }) const result = await applyWorkflowOperations.execute({ @@ -408,6 +413,7 @@ describe('applyWorkflowOperations', () => { reason: 'Block is locked', }, ], + declinedOperationCount: 1, }) const failure = await applyWorkflowOperations @@ -479,6 +485,7 @@ describe('applyWorkflowOperations', () => { state: graph(), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, }) mocks.collectReferences.mockResolvedValue([]) mocks.collectToolReferences.mockResolvedValue([]) @@ -499,6 +506,7 @@ describe('applyWorkflowOperations', () => { state: graph({ 'block-1': { ...BLOCK, locked: true } }), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, }) const result = await applyWorkflowOperations.execute({ @@ -596,6 +604,7 @@ describe('applyWorkflowOperations', () => { }), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, }) const result = await applyWorkflowOperations.execute({ @@ -627,6 +636,7 @@ describe('applyWorkflowOperations', () => { }), validationErrors: [], skippedItems: [], + declinedOperationCount: 0, }) const result = await applyWorkflowOperations.execute({ diff --git a/apps/sim/lib/workflows/application/apply-workflow-operations.ts b/apps/sim/lib/workflows/application/apply-workflow-operations.ts index c3f9e161e51..8595d2ea537 100644 --- a/apps/sim/lib/workflows/application/apply-workflow-operations.ts +++ b/apps/sim/lib/workflows/application/apply-workflow-operations.ts @@ -167,21 +167,6 @@ async function loadStoredGraph(workflowId: string): Promise } -/** - * How many of the engine's skips charge against the operation batch. - * - * The enablement slice appends to the same array, and a deferred forward - * reference is not a refusal at all, so neither may be subtracted from the - * operation count. - */ -function countOperationSkips(skippedItems: readonly SkippedItem[]): number { - let count = 0 - for (const item of skippedItems) { - if (item.operationType !== 'set_block_enabled' && !isDeferredSkippedItem(item)) count += 1 - } - return count -} - /** * How each enablement refusal maps onto the machine-readable skip enum. Kept as * a total `Record` so a new refusal reason fails to compile until it is @@ -296,6 +281,7 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({ state: modifiedGraph, validationErrors, skippedItems, + declinedOperationCount, mintedBlockIds, } = await withBlockVisibility(blockVisibility, async () => applyOperationsToWorkflowState( @@ -308,12 +294,11 @@ export const applyWorkflowOperations = defineAuthorizedWorkflowUseCase({ validationErrors.push(...credentialErrors) /** - * Counted directly rather than as `operations - skipped`. The enablement - * slice pushes its own refusals into the same `skippedItems` array, so - * subtracting the whole array from the operation count charged enablement - * refusals against operations and could go negative. + * Taken before the enablement slice appends its own refusals to `skippedItems`, + * and from the engine's per-operation count: one operation can record several + * skips, so subtracting skips from operations could go negative. */ - const appliedOperations = filteredOperations.length - countOperationSkips(skippedItems) + const appliedOperations = filteredOperations.length - declinedOperationCount const enablement = applyBlockEnabledChanges( modifiedGraph.blocks as Record, diff --git a/apps/sim/lib/workflows/editing/engine.ts b/apps/sim/lib/workflows/editing/engine.ts index 623f4838074..dd3623a388b 100644 --- a/apps/sim/lib/workflows/editing/engine.ts +++ b/apps/sim/lib/workflows/editing/engine.ts @@ -30,7 +30,7 @@ import type { OperationContext, ValidationError, } from './types' -import { logSkippedItem, type SkippedItem } from './types' +import { isDeferredSkippedItem, logSkippedItem, type SkippedItem } from './types' const logger = createLogger('EditWorkflowServerTool') @@ -198,6 +198,7 @@ export function applyOperationsToWorkflowState( deferredConnections: [], } + let declinedOperationCount = 0 for (const operation of orderedOperations) { const { operation_type, block_id } = operation @@ -215,6 +216,7 @@ export function applyOperationsToWorkflowState( block_id, block_id_type: typeof block_id, }) + declinedOperationCount += 1 continue } @@ -226,7 +228,11 @@ export function applyOperationsToWorkflowState( currentBlockCount: Object.keys((modifiedState as any).blocks || {}).length, }) + const skipsBefore = skippedItems.length handler(operation, ctx) + if (skippedItems.slice(skipsBefore).some((item) => !isDeferredSkippedItem(item))) { + declinedOperationCount += 1 + } } // Pass 2: Create all edges from deferred connections @@ -318,6 +324,7 @@ export function applyOperationsToWorkflowState( state: modifiedState, validationErrors, skippedItems, + declinedOperationCount, mintedBlockIds: Object.fromEntries(idMapping), } } diff --git a/apps/sim/lib/workflows/editing/operations.test.ts b/apps/sim/lib/workflows/editing/operations.test.ts index 6ced9063121..1233269d37c 100644 --- a/apps/sim/lib/workflows/editing/operations.test.ts +++ b/apps/sim/lib/workflows/editing/operations.test.ts @@ -1195,6 +1195,40 @@ describe('permission-group tool access', () => { }) ) }) + + it('charges an operation with several skips once, beside other declined operations', () => { + const agentId = '66666666-6666-4666-8666-666666666666' + const cleanId = '77777777-7777-4777-8777-777777777777' + const { skippedItems, declinedOperationCount } = applyOperationsToWorkflowState( + emptyWorkflow(), + [ + { + operation_type: 'add', + block_id: agentId, + params: { + type: 'agent', + name: 'Agent 1', + inputs: { + tools: [ + { type: 'slack', operation: 'canvas', title: 'Canvas A' }, + { type: 'slack', operation: 'canvas', title: 'Canvas B' }, + ], + }, + }, + }, + { + operation_type: 'add', + block_id: cleanId, + params: { type: 'agent', name: 'Agent 2', inputs: {} }, + }, + { operation_type: 'edit', block_id: '', params: {} }, + ], + denyCanvas + ) + + expect(skippedItems.filter((item) => item.type === 'tool_not_allowed')).toHaveLength(2) + expect(declinedOperationCount).toBe(2) + }) }) describe('tool canonical-mode reindexing', () => { diff --git a/apps/sim/lib/workflows/editing/types.ts b/apps/sim/lib/workflows/editing/types.ts index 3ba0e7bcff1..ef27bf741aa 100644 --- a/apps/sim/lib/workflows/editing/types.ts +++ b/apps/sim/lib/workflows/editing/types.ts @@ -154,6 +154,12 @@ export interface ApplyOperationsResult { state: any validationErrors: ValidationError[] skippedItems: SkippedItem[] + /** + * Operations whose handler recorded at least one non-deferred skip. Counted per + * operation, not per skip: one `add` can drop several tools and still be one + * operation, so `operations - skippedItems` would go negative. + */ + declinedOperationCount: number /** * Requested `block_id` -> the id the block was actually given, for every * `add`/`insert_into_subflow` whose requested id was not already a UUID. From aac526240eb2119f1b7eee6ab4adb5906a901750 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 3 Nov 2026 13:08:21 -0800 Subject: [PATCH 2/3] fix(api): count late edge refusals per operation and map MCP list failures to 502 --- .../mothership/integrations/catalog/route.ts | 9 ++++-- apps/sim/lib/workflows/editing/engine.ts | 24 ++++++++++---- .../lib/workflows/editing/operations.test.ts | 32 +++++++++++++++++++ 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/apps/sim/app/api/mothership/integrations/catalog/route.ts b/apps/sim/app/api/mothership/integrations/catalog/route.ts index 570eebbed71..f4779812148 100644 --- a/apps/sim/app/api/mothership/integrations/catalog/route.ts +++ b/apps/sim/app/api/mothership/integrations/catalog/route.ts @@ -5,7 +5,7 @@ import { internalOrchestrationErrorPolicy, internalRateLimits, } from '@/lib/api/server/routes' -import { categorizeMcpServerError } from '@/lib/mcp/utils' +import { categorizeMcpServerError, isMcpServerUnavailableError } from '@/lib/mcp/utils' import { internalCopilotAuth } from '@/lib/mothership/auth/internal' import { INTEGRATION_CATALOG_AUDIENCE, @@ -21,10 +21,15 @@ export const POST = defineInternalJsonRoute({ reason: 'Private bounded catalog discovery rechecks current authorization on every request.', }), errorPolicy: { - /** A targeted MCP server that cannot be reached is the server's failure, not Sim's. */ + /** + * A targeted MCP server that cannot be reached, or that answers `tools/list` with a + * protocol or HTTP error, is the server's failure, not Sim's. + */ project(error) { const mcpFailure = categorizeMcpServerError(error) if (mcpFailure) return internalErrorResponse(mcpFailure.status, { error: mcpFailure.message }) + if (isMcpServerUnavailableError(error)) + return internalErrorResponse(502, { error: 'MCP server did not answer usefully' }) return internalOrchestrationErrorPolicy.project(error) }, unhandled: internalOrchestrationErrorPolicy.unhandled, diff --git a/apps/sim/lib/workflows/editing/engine.ts b/apps/sim/lib/workflows/editing/engine.ts index dd3623a388b..25a9495aee8 100644 --- a/apps/sim/lib/workflows/editing/engine.ts +++ b/apps/sim/lib/workflows/editing/engine.ts @@ -198,8 +198,14 @@ export function applyOperationsToWorkflowState( deferredConnections: [], } - let declinedOperationCount = 0 - for (const operation of orderedOperations) { + /** Indices into `orderedOperations` that recorded a refusal, during their handler or later. */ + const declinedOperations = new Set() + /** The operation that queued each entry of `ctx.deferredConnections`, index for index. */ + const deferredConnectionOrigins: number[] = [] + const recordedRefusalSince = (skipsBefore: number) => + skippedItems.slice(skipsBefore).some((item) => !isDeferredSkippedItem(item)) + + for (const [operationIndex, operation] of orderedOperations.entries()) { const { operation_type, block_id } = operation // CRITICAL: Validate block_id is a valid string and not "undefined" @@ -216,7 +222,7 @@ export function applyOperationsToWorkflowState( block_id, block_id_type: typeof block_id, }) - declinedOperationCount += 1 + declinedOperations.add(operationIndex) continue } @@ -230,8 +236,9 @@ export function applyOperationsToWorkflowState( const skipsBefore = skippedItems.length handler(operation, ctx) - if (skippedItems.slice(skipsBefore).some((item) => !isDeferredSkippedItem(item))) { - declinedOperationCount += 1 + if (recordedRefusalSince(skipsBefore)) declinedOperations.add(operationIndex) + while (deferredConnectionOrigins.length < ctx.deferredConnections.length) { + deferredConnectionOrigins.push(operationIndex) } } @@ -243,7 +250,7 @@ export function applyOperationsToWorkflowState( totalBlocks: Object.keys((modifiedState as any).blocks || {}).length, }) - for (const { blockId, connections } of ctx.deferredConnections) { + for (const [connectionIndex, { blockId, connections }] of ctx.deferredConnections.entries()) { // Verify the source block still exists (it might have been deleted by a later operation) if (!(modifiedState as any).blocks[blockId]) { logger.warn('Source block no longer exists for deferred connection', { @@ -253,6 +260,7 @@ export function applyOperationsToWorkflowState( continue } + const skipsBefore = skippedItems.length addConnectionsAsEdges( modifiedState, blockId, @@ -261,6 +269,8 @@ export function applyOperationsToWorkflowState( skippedItems, validationErrors ) + const origin = deferredConnectionOrigins[connectionIndex] + if (origin !== undefined && recordedRefusalSince(skipsBefore)) declinedOperations.add(origin) } logger.info('Finished processing deferred connections', { @@ -324,7 +334,7 @@ export function applyOperationsToWorkflowState( state: modifiedState, validationErrors, skippedItems, - declinedOperationCount, + declinedOperationCount: declinedOperations.size, mintedBlockIds: Object.fromEntries(idMapping), } } diff --git a/apps/sim/lib/workflows/editing/operations.test.ts b/apps/sim/lib/workflows/editing/operations.test.ts index 1233269d37c..249df5c9f8f 100644 --- a/apps/sim/lib/workflows/editing/operations.test.ts +++ b/apps/sim/lib/workflows/editing/operations.test.ts @@ -1455,6 +1455,38 @@ describe('connection shape validation', () => { }), ]) }) + + it('charges a connection refused after the handler pass to the operation that queued it', () => { + const blockB = '88888888-8888-4888-8888-888888888888' + const { skippedItems, declinedOperationCount } = applyOperationsToWorkflowState( + workflowWithStart(), + [ + { + operation_type: 'add', + block_id: BLOCK_A, + params: { + type: 'function', + name: 'Block A', + inputs: { code: 'return 1' }, + connections: { incoming: 'start-1', outgoing: 'start-1' }, + }, + }, + { + operation_type: 'add', + block_id: blockB, + params: { + type: 'function', + name: 'Block B', + inputs: { code: 'return 2' }, + connections: { success: 'start-1' }, + }, + }, + ] + ) + + expect(skippedItems.filter((item) => item.type === 'invalid_source_handle')).toHaveLength(2) + expect(declinedOperationCount).toBe(1) + }) }) describe('conditional input validation through workflow operations', () => { From 5dd3c8332601f4d5ecd307d7e5348691de3cb4c7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 3 Nov 2026 13:13:49 -0800 Subject: [PATCH 3/3] fix(uploads): leave an already-consumed multipart body as a server error --- apps/sim/lib/core/utils/stream-limits.test.ts | 24 +++++++++++++++++++ apps/sim/lib/core/utils/stream-limits.ts | 10 ++++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/core/utils/stream-limits.test.ts b/apps/sim/lib/core/utils/stream-limits.test.ts index e7eb79beace..6093b586da6 100644 --- a/apps/sim/lib/core/utils/stream-limits.test.ts +++ b/apps/sim/lib/core/utils/stream-limits.test.ts @@ -319,6 +319,30 @@ describe('stream limits', () => { ).rejects.toBeInstanceOf(MultipartFieldValidationError) }) + it('leaves a body that was already read to fail as a server error', async () => { + const input = new FormData() + input.append('name', 'example') + const encoded = new Request('http://localhost/upload', { method: 'POST', body: input }) + const bytes = new Uint8Array(await encoded.arrayBuffer()) + const request = new Request('http://localhost/upload', { + method: 'POST', + headers: { + 'content-type': encoded.headers.get('content-type') ?? '', + 'content-length': String(bytes.byteLength), + }, + body: bytes, + }) + await request.text() + + const failure = await readFormDataWithLimit(request, { + maxBytes: 1024 * 1024, + label: 'multipart body', + }).catch((error: unknown) => error) + + expect(failure).toBeInstanceOf(TypeError) + expect(failure).not.toBeInstanceOf(MultipartFieldValidationError) + }) + it('rejects multipart streams without content-length once bytes exceed the limit', async () => { const request = new Request('http://localhost/upload', { method: 'POST', diff --git a/apps/sim/lib/core/utils/stream-limits.ts b/apps/sim/lib/core/utils/stream-limits.ts index 7587782f69b..e7c96d25b78 100644 --- a/apps/sim/lib/core/utils/stream-limits.ts +++ b/apps/sim/lib/core/utils/stream-limits.ts @@ -69,6 +69,7 @@ export interface ReadFormDataWithLimitRequest { method: string headers?: Headers body?: ReadableStream | null + bodyUsed?: boolean formData: () => Promise } @@ -158,9 +159,14 @@ export async function readFormDataWithLimit( * The fetch body parser rejects a body that does not decode as the content type it * declares (a JSON body, a boundary the body never uses) with a bare `TypeError`. That * is the caller's malformed input, so it surfaces as the same validation failure every - * multipart route already answers with a 400. + * multipart route already answers with a 400. So does a declared-length body that ends + * early, since the client sent an incomplete request. A body some earlier code already + * consumed is a server bug, so that failure is left to surface as it is. */ -async function parseFormData(request: Pick): Promise { +async function parseFormData( + request: Pick +): Promise { + if (request.bodyUsed) return request.formData() try { return await request.formData() } catch (error) {