fix(knowledge): close final ingestion audit gaps

Author: waleedlatif1

apps/sim/background/knowledge-processing.test.ts

@@ -28,7 +28,10 @@ vi.mock('@/lib/knowledge/documents/service', () => ({
 
 import { EmbeddingQuotaExhaustedError } from '@/lib/embeddings/client'
 import { EMBEDDING_QUOTA_CIRCUIT_TTL_MS } from '@/lib/embeddings/quota-circuit'
-import { PermanentDocumentProcessingError } from '@/lib/knowledge/documents/document-processing-error'
+import {
+  PermanentDocumentProcessingError,
+  UsageLimitDocumentProcessingError,
+} from '@/lib/knowledge/documents/document-processing-error'
 import { MAX_QUOTA_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-quota-continuation'
 import {
   resolveQuotaContinuationDelayMs,
@@ -309,6 +312,25 @@ describe('knowledge processing worker', () => {
     })
   })
 
+  it('reports a mutable usage-limit outcome without requesting an immediate retry', async () => {
+    mockProcessDocumentAsync.mockRejectedValue(
+      new UsageLimitDocumentProcessingError('Usage limit exceeded. Upgrade to continue.')
+    )
+
+    await expect(
+      runDocumentProcessing({
+        ...BASE_PAYLOAD,
+        billingScope: 'non-workspace',
+        actorUserId: 'legacy-owner',
+        workspaceId: null,
+      })
+    ).resolves.toMatchObject({
+      success: false,
+      outcome: 'usage_limit',
+      error: 'Usage limit exceeded. Upgrade to continue.',
+    })
+  })
+
   it('preserves normal retries for transient failures', async () => {
     const transientError = new Error('Database connection timed out')
     mockProcessDocumentAsync.mockRejectedValue(transientError)

apps/sim/background/knowledge-processing.ts

@@ -2,7 +2,10 @@ import { createLogger } from '@sim/logger'
 import { task } from '@trigger.dev/sdk'
 import { env, envNumber } from '@/lib/core/config/env'
 import { EMBEDDING_QUOTA_EXHAUSTED_MESSAGE, isEmbeddingQuotaExhaustion } from '@/lib/embeddings'
-import { isPermanentDocumentProcessingError } from '@/lib/knowledge/documents/document-processing-error'
+import {
+  isPermanentDocumentProcessingError,
+  isUsageLimitDocumentProcessingError,
+} from '@/lib/knowledge/documents/document-processing-error'
 import {
   assertDocumentProcessingPayload,
   type DocumentProcessingBillingContext,
@@ -79,6 +82,19 @@ export async function runDocumentProcessing(
       processingTime: Date.now() - startedAt,
     }
   } catch (error) {
+    if (isUsageLimitDocumentProcessingError(error)) {
+      logger.warn(`[${requestId}] Document processing is blocked by the current usage limit`, {
+        filename: docData.filename,
+      })
+      return {
+        success: false,
+        outcome: 'usage_limit' as const,
+        documentId,
+        filename: docData.filename,
+        error: error.message,
+        processingTime: Date.now() - startedAt,
+      }
+    }
     if (isEmbeddingQuotaExhaustion(error)) {
       const outcome = canScheduleQuotaContinuation ? 'quota_deferred' : 'quota_exhausted'
       logger.warn(`[${requestId}] Embedding quota is exhausted`, {
@@ -127,12 +143,8 @@ export const processDocument = task({
     maxTimeoutInMs: envNumber(env.KB_CONFIG_MAX_TIMEOUT, 10000),
     /**
      * `maxAttempts` does not cover an out-of-memory kill — Trigger.dev retries
-     * `TASK_PROCESS_OOM_KILLED` only when a larger preset is named here. Eleven
-     * documents were killed in one afternoon and every one recorded
-     * `attempt_count = 1`, so each was left `failed` with no retry at all. The
-     * escalation is a safety net, not the fix: the workbook parser's allocation
-     * no longer scales with a sheet's declared range, and fleet p99 memory is
-     * 691 MB against this machine's 8 GB.
+     * `TASK_PROCESS_OOM_KILLED` only when a larger preset is named here. The
+     * escalation is a safety net after parser allocations have been bounded.
      */
     outOfMemory: { machine: 'large-2x' },
   },

apps/sim/connectors/fireflies/fireflies.test.ts

@@ -208,6 +208,22 @@ describe('fireflies listDocuments', () => {
     await pending
   })
 
+  it('throws rather than reporting an empty listing when transcripts are missing', async () => {
+    mockGraphQL([{ body: { data: {} } }])
+
+    await expect(firefliesConnector.listDocuments('key', {}, undefined, {})).rejects.toThrow(
+      'Fireflies API returned malformed transcript-list data'
+    )
+  })
+
+  it('rejects malformed transcript rows instead of silently filtering them', async () => {
+    mockGraphQL([{ body: { data: { transcripts: [{}] } } }])
+
+    await expect(firefliesConnector.listDocuments('key', {}, undefined, {})).rejects.toThrow(
+      'Fireflies API returned malformed transcript metadata'
+    )
+  })
+
   it('retries one malformed page without discarding the sync', async () => {
     vi.useFakeTimers()
     mockGraphQL([{ body: {} }, page(2)])
@@ -372,6 +388,17 @@ describe('fireflies getDocument', () => {
     vi.clearAllMocks()
   })
 
+  it.each([null, {}, transcript('different')])(
+    'rejects malformed or mismatched successful transcript metadata',
+    async (value) => {
+      mockGraphQL([{ body: { data: { transcript: value } } }])
+
+      await expect(firefliesConnector.getDocument('key', {}, 't0')).rejects.toThrow(
+        'Fireflies API returned malformed transcript metadata'
+      )
+    }
+  )
+
   it('declares the transcript id as String! and reuses the stub contentHash', async () => {
     const calls = mockGraphQL([
       page(1),

apps/sim/connectors/fireflies/fireflies.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
+import { isPlainRecord } from '@sim/utils/object'
 import { isPayloadSizeLimitError, readResponseTextWithLimit } from '@/lib/core/utils/stream-limits'
 import {
   isRetryableError,
@@ -14,6 +15,7 @@ import {
   ConnectorFileTooLargeError,
   computeContentHash,
   markSkipped,
+  parseOptionalUnlimitedSafeInteger,
   parseTagDate,
 } from '@/connectors/utils'
 
@@ -29,22 +31,11 @@ const FIREFLIES_EXTRACTED_CONTENT_SKIP_REASON =
   'Transcript exceeds the 8MB extracted-content limit and was not indexed'
 const FIREFLIES_RESPONSE_SKIP_REASON =
   'Transcript response exceeds the 16MB safe hydration limit and was not indexed'
+const MAX_TRANSCRIPTS_VALIDATION_ERROR =
+  'Max transcripts must be a positive safe integer, or 0 for unlimited'
 
 function parseMaxTranscripts(value: unknown): number {
-  if (value === undefined || value === null) return 0
-  if (typeof value !== 'string' && typeof value !== 'number') {
-    throw new Error('Max transcripts must be a positive safe integer, or 0 for unlimited')
-  }
-  const normalized = typeof value === 'string' ? value.trim() : value
-  if (normalized === '') return 0
-  if (typeof normalized === 'string' && !/^\d+$/.test(normalized)) {
-    throw new Error('Max transcripts must be a positive safe integer, or 0 for unlimited')
-  }
-  const parsed = Number(normalized)
-  if (!Number.isSafeInteger(parsed) || parsed < 0) {
-    throw new Error('Max transcripts must be a positive safe integer, or 0 for unlimited')
-  }
-  return parsed
+  return parseOptionalUnlimitedSafeInteger(value, MAX_TRANSCRIPTS_VALIDATION_ERROR)
 }
 
 function parsePaginationCursor(cursor?: string): number {
@@ -85,6 +76,19 @@ interface FirefliesTranscript {
   }
 }
 
+function isFirefliesTranscript(value: unknown): value is FirefliesTranscript {
+  return (
+    isPlainRecord(value) &&
+    typeof value.id === 'string' &&
+    value.id.length > 0 &&
+    typeof value.title === 'string' &&
+    typeof value.date === 'number' &&
+    Number.isFinite(value.date) &&
+    typeof value.duration === 'number' &&
+    Number.isFinite(value.duration)
+  )
+}
+
 interface FirefliesGraphQLError {
   message?: string
   code?: string
@@ -459,9 +463,13 @@ export const firefliesConnector: ConnectorConfig = {
       variables
     )
 
-    const transcripts = (
-      Array.isArray(data.transcripts) ? data.transcripts : []
-    ) as FirefliesTranscript[]
+    if (!Array.isArray(data.transcripts)) {
+      throw new Error('Fireflies API returned malformed transcript-list data')
+    }
+    if (!data.transcripts.every(isFirefliesTranscript)) {
+      throw new Error('Fireflies API returned malformed transcript metadata')
+    }
+    const transcripts = data.transcripts as FirefliesTranscript[]
 
     const allStubs = await Promise.all(
       transcripts.filter((t) => Boolean(t?.id)).map(transcriptToStub)
@@ -542,8 +550,10 @@ export const firefliesConnector: ConnectorConfig = {
         { id: externalId }
       )
 
-      const transcript = data.transcript as FirefliesTranscript | null
-      if (!transcript?.id) return null
+      const transcript = data.transcript
+      if (!isFirefliesTranscript(transcript) || transcript.id !== externalId) {
+        throw new Error('Fireflies API returned malformed transcript metadata')
+      }
 
       const stub = await transcriptToStub(transcript)
       let content: string

apps/sim/connectors/google-docs/google-docs.test.ts

@@ -47,6 +47,20 @@ describe('googleDocsConnector', () => {
   })
 
   describe('getDocument', () => {
+    it.each([{}, { ...DRIVE_FILE, id: 'different-document' }])(
+      'rejects malformed Drive metadata instead of replacing retained content',
+      async (metadata) => {
+        vi.stubGlobal(
+          'fetch',
+          vi.fn(async () => new Response(JSON.stringify(metadata), { status: 200 }))
+        )
+
+        await expect(
+          googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)
+        ).rejects.toThrow('Google Drive API returned malformed file metadata')
+      }
+    )
+
     it('authoritatively skips a listed document that changed type', async () => {
       vi.stubGlobal(
         'fetch',
@@ -257,6 +271,42 @@ describe('googleDocsConnector', () => {
       })
     })
 
+    it('rejects a successful response that omits the required tabs array', async () => {
+      stubFetchDocument(new Response('{}', { status: 200 }))
+
+      await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
+        'Google Docs API returned a malformed document response'
+      )
+    })
+
+    it('rejects an empty tab object instead of replacing retained content', async () => {
+      stubFetchDocument(new Response(JSON.stringify({ tabs: [{}] }), { status: 200 }))
+
+      await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
+        'Google Docs API returned a malformed document response'
+      )
+    })
+
+    it('rejects malformed nested tab content instead of dropping it', async () => {
+      stubFetchDocument(
+        new Response(
+          JSON.stringify({
+            tabs: [
+              {
+                documentTab: { body: { content: [] } },
+                childTabs: [{}],
+              },
+            ],
+          }),
+          { status: 200 }
+        )
+      )
+
+      await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
+        'Google Docs API returned a malformed document response'
+      )
+    })
+
     it('maps an oversized chunked hydration response to a visible skip', async () => {
       const chunk = new Uint8Array(34 * 1024 * 1024)
       let chunksSent = 0
@@ -473,6 +523,46 @@ describe('googleDocsConnector', () => {
   })
 
   describe('listDocuments', () => {
+    it('rejects a malformed successful Drive list envelope', async () => {
+      vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response('{}', { status: 200 })))
+
+      await expect(
+        googleDocsConnector.listDocuments(ACCESS_TOKEN, {}, undefined, {})
+      ).rejects.toThrow('Google Drive API returned malformed file-list metadata')
+    })
+
+    it('accepts a discriminator-only empty Drive list envelope', async () => {
+      vi.stubGlobal(
+        'fetch',
+        vi
+          .fn()
+          .mockResolvedValue(
+            new Response(JSON.stringify({ kind: 'drive#fileList' }), { status: 200 })
+          )
+      )
+
+      await expect(
+        googleDocsConnector.listDocuments(ACCESS_TOKEN, {}, undefined, {})
+      ).resolves.toMatchObject({ documents: [], hasMore: false })
+    })
+
+    it.each([
+      {
+        files: [{ name: 'Missing ID', mimeType: DRIVE_FILE.mimeType, modifiedTime: '2026-01-01' }],
+      },
+      { files: [], nextPageToken: 123 },
+      { files: [], incompleteSearch: 'true' },
+    ])('rejects malformed Drive list metadata', async (body) => {
+      vi.stubGlobal(
+        'fetch',
+        vi.fn().mockResolvedValue(new Response(JSON.stringify(body), { status: 200 }))
+      )
+
+      await expect(
+        googleDocsConnector.listDocuments(ACCESS_TOKEN, {}, undefined, {})
+      ).rejects.toThrow('Google Drive API returned malformed file-list metadata')
+    })
+
     it('does not issue another Drive request after a lowered cap is already exhausted', async () => {
       const fetchMock = vi.fn()
       vi.stubGlobal('fetch', fetchMock)
@@ -586,9 +676,7 @@ describe('googleDocsConnector', () => {
       async (maxDocs) => {
         const fetchMock = vi
           .fn()
-          .mockResolvedValue(
-            new Response(JSON.stringify({ files: [], nextPageToken: null }), { status: 200 })
-          )
+          .mockResolvedValue(new Response(JSON.stringify({ files: [] }), { status: 200 }))
         vi.stubGlobal('fetch', fetchMock)
 
         await expect(