fix(knowledge): parse the stored artifact, not the document's display name (#6817)

* fix(knowledge): parse the stored artifact, not the document's display name

A connector document's `filename` is a display name that deliberately disagrees
with the bytes on disk: the sync engine records the source file's name
(`Report.pdf`) while storing the text the connector already extracted from it
under a `.txt` key, with `mimeType: 'text/plain'`.

`processDocumentAsync` discards the processing filename the sync engine computes
and rebuilds its input from the document row, so the parser was chosen from the
display name and re-parsed extracted text as the source binary. In production
that failed 1,379 SharePoint PDFs with `Invalid PDF structure.` and silently
double-wrapped 364 spreadsheets — those reported `completed`, wrapping a second
fake sheet around the connector's own extraction, because SheetJS accepts almost
any input.

Parser selection now prefers the extension of the object actually fetched,
falling back to the filename/MIME path when the URL is not ours or the key
carries no extension a parser claims. Both ingestion paths are honest under that
rule because `fitStorageKeyName` preserves extensions through truncation: an
upload keys on its original name, a connector document keys on what it stored.

This layer is what covers the stuck-document retry sweep, which rebuilds its own
input from the same display name — the sweep is the path that reprocesses the
already-failed documents, so a fix confined to `processDocumentAsync` would have
left the remediation itself broken.

The defect predates the connectors that expose it: Box fetches Box-side text
representations for `pdf`/`docx`/`xlsx` and stores them under the source name
too, so it was latent there before SharePoint and OneDrive reached binary
formats.

`connectorArtifactFileName` now owns the `.txt` suffix that the parser choice
depends on, so the invariant is structural instead of a convention repeated at
four call sites per function.

* fix(knowledge): raise the connector sync ceiling and tie it to the stale lock

A 2,600-document library exhausted the 30-minute budget and the run was killed
mid-listing, leaving the connector's `syncing` lock set until the scheduler
reclaimed it.

Raising the ceiling is not a lone constant, because reclaiming a stale lock
flips the connector to `error` and frees it for another sync. A TTL at or below
the run ceiling would hand the lock to a successor while the first sync is still
writing — two syncs racing the same `(connectorId, externalId)` rows. The
previous values, a 1800s run against a hard-coded 120-minute TTL declared in a
different file, held that invariant only by coincidence.

Both now derive from one another, with a test pinning the margin so the next
raise cannot silently break it.

Author: waleedlatif1

apps/sim/app/api/knowledge/connectors/sync/route.ts

@@ -9,6 +9,7 @@ import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { dispatchSync } from '@/lib/knowledge/connectors/queue'
+import { CONNECTOR_SYNC_STALE_LOCK_TTL_MS } from '@/lib/knowledge/connectors/sync-limits'
 
 export const dynamic = 'force-dynamic'
 
@@ -39,8 +40,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const now = new Date()
 
-    const STALE_SYNC_TTL_MS = 120 * 60 * 1000
-    const staleCutoff = new Date(now.getTime() - STALE_SYNC_TTL_MS)
+    const staleCutoff = new Date(now.getTime() - CONNECTOR_SYNC_STALE_LOCK_TTL_MS)
 
     const recoveredConnectors = await db
       .update(knowledgeConnector)

apps/sim/background/knowledge-connector-sync.ts

@@ -5,6 +5,7 @@ import {
   type ConnectorSyncPayload,
 } from '@/lib/knowledge/connectors/queue'
 import { executeSync } from '@/lib/knowledge/connectors/sync-engine'
+import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/sync-limits'
 
 const logger = createLogger('TriggerKnowledgeConnectorSync')
 
@@ -39,7 +40,7 @@ export async function executeConnectorSyncJob(payload: unknown) {
 
 export const knowledgeConnectorSync = task({
   id: 'knowledge-connector-sync',
-  maxDuration: 1800,
+  maxDuration: CONNECTOR_SYNC_MAX_DURATION_SECONDS,
   machine: 'large-2x',
   retry: {
     maxAttempts: 3,

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -64,6 +64,20 @@ const MAX_CONSECUTIVE_FAILURES = 10
 function sanitizeStorageTitle(title: string): string {
   return title.replace(/[^a-zA-Z0-9.-]/g, '_').slice(0, MAX_SAFE_TITLE_LENGTH)
 }
+
+/**
+ * Name a connector document's stored object carries.
+ *
+ * Connectors store already-extracted text while `document.filename` keeps the
+ * source file's name for display, so the stored object has to declare the format
+ * it actually holds: `resolveStoredArtifactExtension` picks the parser off this
+ * key, and a key ending in the source extension would re-parse extracted text as
+ * the original binary. Owning the `.txt` suffix here makes that structural rather
+ * than a convention each call site has to remember.
+ */
+function connectorArtifactFileName(title: string): string {
+  return `${sanitizeStorageTitle(title)}.txt`
+}
 type KnowledgeBaseLockingTx = Pick<typeof db, 'execute' | 'select'>
 
 type DocOp =
@@ -1657,17 +1671,17 @@ async function addDocument(
 ): Promise<DocumentData> {
   const documentId = generateId()
   const contentBuffer = Buffer.from(extDoc.content, 'utf-8')
-  const safeTitle = sanitizeStorageTitle(extDoc.title)
-  const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${documentId}-`, `${safeTitle}.txt`)}`
+  const storedFileName = connectorArtifactFileName(extDoc.title)
+  const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${documentId}-`, storedFileName)}`
 
   const fileInfo = await StorageService.uploadFile({
     file: contentBuffer,
-    fileName: `${safeTitle}.txt`,
+    fileName: storedFileName,
     contentType: 'text/plain',
     context: 'knowledge-base',
     customKey,
     preserveKey: true,
-    metadata: kbOwnershipMetadata(kbOwner, `${safeTitle}.txt`),
+    metadata: kbOwnershipMetadata(kbOwner, storedFileName),
   })
 
   const fileUrl = `${getInternalApiBaseUrl()}${fileInfo.path}?context=knowledge-base`
@@ -1676,8 +1690,6 @@ async function addDocument(
     ? resolveTagMapping(connectorType, extDoc.metadata, sourceConfig)
     : undefined
 
-  const processingFilename = `${safeTitle}.txt`
-
   try {
     await db.transaction(async (tx) => {
       const isActive = await isKnowledgeBaseActiveInTx(tx, knowledgeBaseId)
@@ -1718,7 +1730,7 @@ async function addDocument(
 
   return {
     documentId,
-    filename: processingFilename,
+    filename: storedFileName,
     fileUrl,
     fileSize: contentBuffer.length,
     mimeType: 'text/plain',
@@ -1746,17 +1758,17 @@ async function updateDocument(
   const oldFileUrl = existingRows[0]?.fileUrl
 
   const contentBuffer = Buffer.from(extDoc.content, 'utf-8')
-  const safeTitle = sanitizeStorageTitle(extDoc.title)
-  const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${existingDocId}-`, `${safeTitle}.txt`)}`
+  const storedFileName = connectorArtifactFileName(extDoc.title)
+  const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${existingDocId}-`, storedFileName)}`
 
   const fileInfo = await StorageService.uploadFile({
     file: contentBuffer,
-    fileName: `${safeTitle}.txt`,
+    fileName: storedFileName,
     contentType: 'text/plain',
     context: 'knowledge-base',
     customKey,
     preserveKey: true,
-    metadata: kbOwnershipMetadata(kbOwner, `${safeTitle}.txt`),
+    metadata: kbOwnershipMetadata(kbOwner, storedFileName),
   })
 
   const fileUrl = `${getInternalApiBaseUrl()}${fileInfo.path}?context=knowledge-base`
@@ -1765,8 +1777,6 @@ async function updateDocument(
     ? resolveTagMapping(connectorType, extDoc.metadata, sourceConfig)
     : undefined
 
-  const processingFilename = `${safeTitle}.txt`
-
   try {
     await db.transaction(async (tx) => {
       const isActive = await isKnowledgeBaseActiveInTx(tx, knowledgeBaseId)
@@ -1839,7 +1849,7 @@ async function updateDocument(
 
   return {
     documentId: existingDocId,
-    filename: processingFilename,
+    filename: storedFileName,
     fileUrl,
     fileSize: contentBuffer.length,
     mimeType: 'text/plain',

apps/sim/lib/knowledge/connectors/sync-limits.test.ts

@@ -0,0 +1,26 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import {
+  CONNECTOR_SYNC_MAX_DURATION_SECONDS,
+  CONNECTOR_SYNC_STALE_LOCK_TTL_MS,
+} from '@/lib/knowledge/connectors/sync-limits'
+
+describe('connector sync limits', () => {
+  /**
+   * Reclaiming a stale lock frees it for another sync, so a TTL at or below the
+   * run ceiling would start a second sync while the first is still writing. This
+   * guards the invariant against a future hard-coded TTL, not the derivation.
+   */
+  it('keeps at least a 2x margin between the run ceiling and the reclaim', () => {
+    expect(CONNECTOR_SYNC_STALE_LOCK_TTL_MS).toBeGreaterThanOrEqual(
+      CONNECTOR_SYNC_MAX_DURATION_SECONDS * 2 * 1000
+    )
+  })
+
+  /** A 2,600-document library exhausted the previous 1800s budget mid-listing. */
+  it('allows a run longer than the half hour that timed out in production', () => {
+    expect(CONNECTOR_SYNC_MAX_DURATION_SECONDS).toBeGreaterThan(1800)
+  })
+})

apps/sim/lib/knowledge/connectors/sync-limits.ts

@@ -0,0 +1,16 @@
+/**
+ * Wall-clock ceiling for a single connector sync run. A large document library
+ * needs more than the half hour this used to allow: a 2,600-document site
+ * exhausted the old budget and was killed mid-listing, leaving its `syncing`
+ * lock set until the scheduler reclaimed it.
+ */
+export const CONNECTOR_SYNC_MAX_DURATION_SECONDS = 3600
+
+/**
+ * How long a connector may sit in `syncing` before the scheduler reclaims its lock.
+ *
+ * MUST stay above {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS}: reclaiming frees the
+ * lock for another sync, so a TTL at or below the run ceiling would start a second
+ * sync while the first is still writing, both racing the same documents.
+ */
+export const CONNECTOR_SYNC_STALE_LOCK_TTL_MS = CONNECTOR_SYNC_MAX_DURATION_SECONDS * 2 * 1000

apps/sim/lib/knowledge/documents/document-processor.ts

@@ -18,7 +18,10 @@ import { env, envNumber } from '@/lib/core/config/env'
 import { OCR_CAPABILITY, requireCapability } from '@/lib/core/config/env-capabilities'
 import { parseBuffer } from '@/lib/file-parsers'
 import type { FileParseMetadata } from '@/lib/file-parsers/types'
-import { resolveParserExtension } from '@/lib/knowledge/documents/parser-extension'
+import {
+  resolveParserExtension,
+  resolveStoredArtifactExtension,
+} from '@/lib/knowledge/documents/parser-extension'
 import { retryWithExponentialBackoff } from '@/lib/knowledge/documents/utils'
 import {
   assertKnowledgeOpaqueModelInputSafe,
@@ -841,7 +844,9 @@ async function parseHttpFile(
 ): Promise<{ content: string; metadata?: FileParseMetadata }> {
   const buffer = await downloadFileWithTimeout(fileUrl, userId)
 
-  const extension = resolveParserExtension(filename, mimeType)
+  /** Prefer what we actually downloaded over what the document is *called*. */
+  const extension =
+    resolveStoredArtifactExtension(fileUrl) ?? resolveParserExtension(filename, mimeType)
   const result = await parseBuffer(buffer, extension)
   return result
 }

apps/sim/lib/knowledge/documents/parser-extension.ts

@@ -1,4 +1,9 @@
-import { getExtensionFromMimeType } from '@/lib/uploads/utils/file-utils'
+import {
+  extractStorageKey,
+  getExtensionFromMimeType,
+  getFileExtension,
+  isInternalFileUrl,
+} from '@/lib/uploads/utils/file-utils'
 import {
   isAlphanumericExtension,
   isSupportedExtension,
@@ -12,8 +17,8 @@ export function resolveParserExtension(
   mimeType?: string,
   fallback?: string
 ): string {
-  const raw = filename.includes('.') ? filename.split('.').pop()?.toLowerCase() : undefined
-  const filenameExtension = raw && isAlphanumericExtension(raw) ? raw : undefined
+  const raw = getFileExtension(filename)
+  const filenameExtension = isAlphanumericExtension(raw) ? raw : undefined
 
   if (filenameExtension && isSupportedExtension(filenameExtension)) {
     return filenameExtension
@@ -36,3 +41,31 @@ export function resolveParserExtension(
 
   throw new Error(`Could not determine file type for ${filename || 'document'}`)
 }
+
+/**
+ * Extension of the object actually stored, taken from its storage key.
+ *
+ * A knowledge base document's `filename` is a *display* name, and for connector
+ * documents it deliberately disagrees with the bytes on disk: the sync engine
+ * records the source file's name (`Report.pdf`) while storing the text the
+ * connector already extracted from it under a `.txt` key. Choosing a parser from
+ * the display name therefore re-parses extracted text as the original binary
+ * format — `Invalid PDF structure.` for PDFs, and for spreadsheets a silent
+ * double-wrap, since SheetJS accepts almost anything.
+ *
+ * The storage key is the honest signal for both ingestion paths, because
+ * `fitStorageKeyName` preserves a file's extension through truncation: an upload
+ * keys on its original name (`kb/<id>-Report.pdf`) and a connector document keys
+ * on what it stored (`kb/<id>-Report.pdf.txt`).
+ *
+ * Falls back to `undefined` — leaving the caller on the filename/MIME path —
+ * rather than guessing, so this can only ever redirect to a parser that exists.
+ */
+export function resolveStoredArtifactExtension(fileUrl: string): string | undefined {
+  if (!isInternalFileUrl(fileUrl)) return undefined
+
+  const extension = getFileExtension(extractStorageKey(fileUrl))
+  if (!isAlphanumericExtension(extension)) return undefined
+
+  return isSupportedExtension(extension) ? extension : undefined
+}

apps/sim/lib/knowledge/documents/stored-artifact-extension.test.ts

@@ -0,0 +1,73 @@
+/**
+ * @vitest-environment node
+ *
+ * A knowledge base document's `filename` is a display name. For connector
+ * documents it deliberately disagrees with the stored bytes — the sync engine
+ * records `Report.pdf` while storing the text the connector already extracted
+ * under a `.txt` key — so choosing a parser from the display name re-parsed
+ * extracted text as the source binary. In production that failed 1,379
+ * SharePoint PDFs with `Invalid PDF structure.` and silently double-wrapped
+ * every spreadsheet, which "succeeded" because SheetJS accepts almost anything.
+ */
+import { describe, expect, it } from 'vitest'
+import { resolveStoredArtifactExtension } from '@/lib/knowledge/documents/parser-extension'
+
+const CONNECTOR_PDF_URL =
+  '/api/files/serve/s3/kb%2F1786986883507-abc-Report.pdf.txt?context=knowledge-base'
+const UPLOADED_PDF_URL =
+  '/api/files/serve/s3/kb%2F1786986883507-abc-Report.pdf?context=knowledge-base'
+
+describe('resolveStoredArtifactExtension', () => {
+  it('reports txt for a connector document whose display name is a PDF', () => {
+    expect(resolveStoredArtifactExtension(CONNECTOR_PDF_URL)).toBe('txt')
+  })
+
+  it('reports txt for a connector spreadsheet, which SheetJS would otherwise re-wrap', () => {
+    expect(
+      resolveStoredArtifactExtension(
+        '/api/files/serve/s3/kb%2F1-abc-Vendor_Spend.xlsx.txt?context=knowledge-base'
+      )
+    ).toBe('txt')
+  })
+
+  it('leaves an uploaded document on its real extension', () => {
+    expect(resolveStoredArtifactExtension(UPLOADED_PDF_URL)).toBe('pdf')
+  })
+
+  it('handles the blob and gcs storage prefixes', () => {
+    expect(resolveStoredArtifactExtension('/api/files/serve/blob/kb%2F1-a-x.docx')).toBe('docx')
+    expect(resolveStoredArtifactExtension('/api/files/serve/gcs/kb%2F1-a-x.csv')).toBe('csv')
+  })
+
+  it('ignores URLs that are not served from our own storage', () => {
+    expect(resolveStoredArtifactExtension('https://example.com/files/Report.pdf')).toBeUndefined()
+    expect(resolveStoredArtifactExtension('data:application/pdf;base64,AAAA')).toBeUndefined()
+  })
+
+  /**
+   * `fitStorageKeyName` drops the extension when it cannot fit, and a key may
+   * carry no extension at all. Returning undefined puts the caller back on the
+   * filename/MIME path rather than guessing.
+   */
+  it('returns undefined when the key carries no usable extension', () => {
+    expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report')).toBeUndefined()
+    expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.')).toBeUndefined()
+  })
+
+  /**
+   * Only ever redirects to a parser that exists — an unknown suffix falls back
+   * instead of routing the document at a parser that cannot handle it.
+   */
+  it('returns undefined for an extension no parser claims', () => {
+    expect(
+      resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-archive.zip')
+    ).toBeUndefined()
+    expect(
+      resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.v2.final')
+    ).toBeUndefined()
+  })
+
+  it('is case-insensitive', () => {
+    expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.PDF')).toBe('pdf')
+  })
+})