@@ -204,6 +204,25 @@ describe('googleDocsConnector', () => {
       expect(error?.message).not.toContain(secret)
     })
 
+    it('returns only the status when an error envelope exceeds the diagnostic limit', async () => {
+      const secret = 'over-limit-provider-secret-that-must-not-escape'
+      stubFetchDocument(
+        new Response(`${'x'.repeat(64 * 1024)}${secret}`, {
+          status: 400,
+          headers: { 'Content-Type': 'application/json' },
+        })
+      )
+
+      const error = await googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID).then(
+        () => undefined,
+        (caught) => caught as Error
+      )
+
+      expect(error?.message).toBe(`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400`)
+      expect(error?.message).not.toContain('response body omitted')
+      expect(error?.message).not.toContain(secret)
+    })
+
     it('marks an empty tab response as an authoritative skip', async () => {
       stubFetchDocument(new Response(JSON.stringify({ tabs: [] }), { status: 200 }))
 
 
@@ -149,7 +149,10 @@ interface DocsDocument {
 
 /** Describes a Google API failure without leaking an unbounded response body. */
 async function describeGoogleApiFailure(response: Response): Promise<string> {
-  const rawBody = await readBoundedHttpErrorPayload(response)
+  const payload = await readBoundedHttpErrorPayload(response)
+  if (!payload.ok) return String(response.status)
+
+  const rawBody = payload.body
   const normalizedBody = sanitizeHttpErrorDiagnostic(rawBody).replace(/\s+/g, ' ').trim()
   if (!normalizedBody) return String(response.status)
 
 
@@ -3,12 +3,14 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockFetchWithRetry } = vi.hoisted(() => ({ mockFetchWithRetry: vi.fn() }))
+const { mockFetchWithRetry, mockReadBoundedHttpErrorPayload } = vi.hoisted(() => ({
+  mockFetchWithRetry: vi.fn(),
+  mockReadBoundedHttpErrorPayload: vi.fn(),
+}))
 
 vi.mock('@/lib/knowledge/documents/utils', () => ({
   fetchWithRetry: mockFetchWithRetry,
-  readBoundedHttpErrorBody: async (response: Response) => (await response.text()).slice(0, 2000),
-  readBoundedHttpErrorPayload: async (response: Response) => response.text(),
+  readBoundedHttpErrorPayload: mockReadBoundedHttpErrorPayload,
   VALIDATE_RETRY_OPTIONS: {},
 }))
 vi.mock('@/components/icons', () => ({ NotionIcon: () => null }))
@@ -52,6 +54,14 @@ function dataSourceCursor(value: Record<string, unknown>): string {
   return `notion-data-sources:v1:${encodeURIComponent(JSON.stringify(value))}`
 }
 
+beforeEach(() => {
+  mockReadBoundedHttpErrorPayload.mockReset()
+  mockReadBoundedHttpErrorPayload.mockImplementation(async (response: Response) => ({
+    ok: true,
+    body: await response.text(),
+  }))
+})
+
 describe('notion markdown hydration', () => {
   beforeEach(() => {
     mockFetchWithRetry.mockReset()
@@ -226,6 +236,35 @@ describe('notion markdown hydration', () => {
     )
   })
 
+  it('recovers the expression from an unsupported equation block', async () => {
+    mockFetchWithRetry
+      .mockResolvedValueOnce(notionResponse(page()))
+      .mockResolvedValueOnce(
+        notionResponse({
+          markdown: 'Before unsupported equation: <unknown>',
+          truncated: false,
+          unknown_block_ids: ['equation-1'],
+        })
+      )
+      .mockResolvedValueOnce(
+        notionResponse({
+          object: 'block',
+          id: 'equation-1',
+          type: 'equation',
+          equation: { expression: 'e=mc^2' },
+        })
+      )
+
+    const document = await notionConnector.getDocument('token', {}, 'page-1')
+
+    expect(document).toMatchObject({ contentDeferred: false })
+    expect(document?.content).toContain('e=mc^2')
+    expect(document?.skippedReason).toBeUndefined()
+    expect(String(mockFetchWithRetry.mock.calls[2][0])).toBe(
+      'https://api.notion.com/v1/blocks/equation-1'
+    )
+  })
+
   it('marks an unrecoverable truncated response as skipped rather than storing partial content', async () => {
     mockFetchWithRetry
       .mockResolvedValueOnce(notionResponse(page()))
@@ -340,6 +379,29 @@ describe('notion markdown hydration', () => {
     )
   })
 
+  it('degrades to status-only diagnostics when the bounded error payload is unavailable', async () => {
+    mockReadBoundedHttpErrorPayload.mockResolvedValueOnce({ ok: false, reason: 'too_large' })
+    mockFetchWithRetry.mockResolvedValueOnce(notionResponse(page())).mockResolvedValueOnce(
+      notionResponse(
+        {
+          object: 'error',
+          code: 'validation_error',
+          message: 'This diagnostic must not be consumed',
+          request_id: 'request-omitted',
+        },
+        400
+      )
+    )
+
+    await expect(notionConnector.getDocument('token', {}, 'page-1')).rejects.toMatchObject({
+      message: 'Failed to fetch markdown for page-1: 400',
+      status: 400,
+      code: undefined,
+      requestId: undefined,
+    })
+    expect(mockReadBoundedHttpErrorPayload).toHaveBeenCalledTimes(1)
+  })
+
   it('records an oversized markdown response as an intrinsic skipped document', async () => {
     mockFetchWithRetry.mockResolvedValueOnce(notionResponse(page())).mockResolvedValueOnce(
       new Response('{}', {
@@ -473,6 +535,49 @@ describe('notion listing completeness', () => {
     expect(mockFetchWithRetry).toHaveBeenCalledTimes(3)
   })
 
+  it.each([undefined, null, '', '   '])(
+    'stops safely when a data source has more rows without a usable cursor: %j',
+    async (nextCursor) => {
+      mockFetchWithRetry
+        .mockResolvedValueOnce(
+          notionResponse({
+            object: 'database',
+            id: 'database-1',
+            data_sources: [
+              { id: 'source-1', name: 'Primary' },
+              { id: 'source-2', name: 'Archive' },
+            ],
+          })
+        )
+        .mockResolvedValueOnce(
+          notionResponse({
+            results: [page('page-1')],
+            has_more: true,
+            next_cursor: nextCursor,
+          })
+        )
+
+      const syncContext: Record<string, unknown> = {}
+      const result = await notionConnector.listDocuments(
+        'token',
+        { scope: 'database', databaseId: 'database-1' },
+        undefined,
+        syncContext
+      )
+
+      expect(result.documents.map((document) => document.externalId)).toEqual(['page-1'])
+      expect(result.hasMore).toBe(false)
+      expect(result.nextCursor).toBeUndefined()
+      expect(result.reconciliationSafe).toBe(false)
+      expect(syncContext.listingCapped).toBe(true)
+      expect(syncContext.reconciliationUnsafe).toBe(true)
+      expect(mockFetchWithRetry).toHaveBeenCalledTimes(2)
+      expect(
+        mockFetchWithRetry.mock.calls.some(([url]) => String(url).includes('/source-2/query'))
+      ).toBe(false)
+    }
+  )
+
   it('makes a data-source listing non-authoritative when Notion reports its 10,000-row ceiling', async () => {
     mockFetchWithRetry
       .mockResolvedValueOnce(
 
@@ -175,9 +175,12 @@ async function notionApiError(response: Response, operation: string): Promise<No
   let body: NotionApiErrorBody = {}
 
   try {
-    const parsed: unknown = JSON.parse(await readBoundedHttpErrorPayload(response))
-    if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
-      body = parsed as NotionApiErrorBody
+    const payload = await readBoundedHttpErrorPayload(response)
+    if (payload.ok) {
+      const parsed: unknown = JSON.parse(payload.body)
+      if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
+        body = parsed as NotionApiErrorBody
+      }
     }
   } catch {
     body = {}
@@ -292,10 +295,13 @@ function unsupportedBlockFallback(block: Record<string, unknown>): string {
   if (!payload || Array.isArray(payload)) return ''
 
   const value = payload as Record<string, unknown>
+  const expression =
+    type === 'equation' && typeof value.expression === 'string' ? value.expression : ''
   const text =
     richTextToPlainText(value.rich_text) ||
     richTextToPlainText(value.caption) ||
-    richTextToPlainText(value.title)
+    richTextToPlainText(value.title) ||
+    expression
   const url = typeof value.url === 'string' ? value.url : ''
   return [text, url].filter(Boolean).join('\n')
 }
@@ -1032,20 +1038,26 @@ async function listFromDatabases(
     queryResultIncomplete =
       data.request_status?.type === 'incomplete' &&
       data.request_status?.incomplete_reason === 'query_result_limit_reached'
+    const providerCursor =
+      typeof data.next_cursor === 'string' && data.next_cursor.trim().length > 0
+        ? data.next_cursor
+        : undefined
+    const paginationCursorMissing = data.has_more === true && providerCursor === undefined
 
-    if (queryResultIncomplete && syncContext) {
+    if ((queryResultIncomplete || paginationCursorMissing) && syncContext) {
       syncContext.listingCapped = true
       syncContext.reconciliationUnsafe = true
     }
 
-    if (data.has_more === true && typeof data.next_cursor === 'string') {
-      const nextStart = data.next_cursor as string
-      nextCursor = encodeDataSourceCursor({ sourceIndex, cursor: nextStart })
+    if (data.has_more === true && providerCursor !== undefined) {
+      nextCursor = encodeDataSourceCursor({ sourceIndex, cursor: providerCursor })
       hasMore = true
-    } else if (sourceIndex + 1 < dataSources.length) {
+    } else if (!paginationCursorMissing && sourceIndex + 1 < dataSources.length) {
       nextCursor = encodeDataSourceCursor({ sourceIndex: sourceIndex + 1 })
       hasMore = true
     }
+
+    if (paginationCursorMissing) queryResultIncomplete = true
   }
 
   const totalFetched = ((syncContext?.totalDocsFetched as number) ?? 0) + documents.length
 
@@ -811,6 +811,145 @@ describe('processDocumentsWithQueue dispatch backend', () => {
     ).toBe(false)
   })
 
+  it('withdraws a direct dispatch whose guarded processing claim lost the race', async () => {
+    setEnvFlags({ isTriggerDevEnabled: true })
+    dbChainMockFns.returning.mockReset()
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'document-1' }]).mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockReset()
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([{ userId: 'knowledge-owner', workspaceId: 'workspace-1' }])
+      .mockResolvedValueOnce([
+        {
+          knowledgeBaseUserId: 'knowledge-owner',
+          workspaceId: 'workspace-1',
+          filename: DOCUMENT.filename,
+          fileUrl: DOCUMENT.fileUrl,
+          fileSize: DOCUMENT.fileSize,
+          mimeType: DOCUMENT.mimeType,
+        },
+      ])
+      .mockResolvedValueOnce([])
+
+    await expect(
+      processDocumentsWithQueue(
+        [DOCUMENT],
+        'knowledge-base-1',
+        {},
+        'request-1',
+        BILLING_ATTRIBUTION
+      )
+    ).rejects.toThrow('document processing dispatches failed')
+
+    expect(mockBatchTrigger).not.toHaveBeenCalled()
+    expect(
+      dbChainMockFns.set.mock.calls.some(
+        (call) =>
+          (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt === null &&
+          'processingAttempts' in ((call[0] as Record<string, unknown> | undefined) ?? {})
+      )
+    ).toBe(true)
+  })
+
+  it('accepts an unclaimed direct dispatch only after revalidating live document state', async () => {
+    vi.useFakeTimers()
+    const now = new Date('2026-08-25T06:00:00.000Z')
+    vi.setSystemTime(now)
+    setEnvFlags({ isTriggerDevEnabled: true })
+    dbChainMockFns.returning.mockReset()
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'document-1' }]).mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockReset()
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([{ userId: 'knowledge-owner', workspaceId: 'workspace-1' }])
+      .mockResolvedValueOnce([
+        {
+          knowledgeBaseUserId: 'knowledge-owner',
+          workspaceId: 'workspace-1',
+          filename: DOCUMENT.filename,
+          fileUrl: DOCUMENT.fileUrl,
+          fileSize: DOCUMENT.fileSize,
+          mimeType: DOCUMENT.mimeType,
+        },
+      ])
+      .mockResolvedValueOnce([{ id: 'document-1' }])
+
+    await expect(
+      processDocumentsWithQueue(
+        [DOCUMENT],
+        'knowledge-base-1',
+        {},
+        'request-1',
+        BILLING_ATTRIBUTION
+      )
+    ).resolves.toEqual({ requested: 1, accepted: 1, failed: 0, failedDocumentIds: [] })
+
+    const revalidationGuard = dbChainMockFns.where.mock.calls.find(
+      (call) =>
+        hasMockCondition(
+          call[0],
+          (node: MockCondition) =>
+            node.type === 'eq' &&
+            node.left === schemaMock.document.id &&
+            node.right === 'document-1'
+        ) &&
+        flattenMockConditions(call[0]).some(
+          (node: MockCondition) =>
+            node.type === 'or' &&
+            (node.conditions as MockCondition[]).some(
+              (condition) =>
+                condition.type === 'eq' &&
+                condition.left === schemaMock.document.processingStatus &&
+                condition.right === 'completed'
+            )
+        )
+    )?.[0]
+    expect(revalidationGuard).toBeDefined()
+    const acceptedStatusGuard = flattenMockConditions(revalidationGuard).find(
+      (node: MockCondition) => node.type === 'or'
+    )
+    const acceptedStatuses = acceptedStatusGuard?.conditions as MockCondition[]
+    const pendingState = acceptedStatuses.find(
+      (condition) =>
+        condition.type === 'and' &&
+        hasMockCondition(
+          condition,
+          (node: MockCondition) =>
+            node.type === 'eq' &&
+            node.left === schemaMock.document.processingStatus &&
+            node.right === 'pending'
+        )
+    )
+    const queuedFreshness = flattenMockConditions(pendingState).find(
+      (node: MockCondition) =>
+        node.type === 'gte' && node.left === schemaMock.document.processingQueuedAt
+    )
+    expect(queuedFreshness?.right).toEqual(new Date(now.getTime() - QUEUED_DISPATCH_GRACE_MS))
+    const processingState = acceptedStatuses.find(
+      (condition) =>
+        condition.type === 'and' &&
+        hasMockCondition(
+          condition,
+          (node: MockCondition) =>
+            node.type === 'eq' &&
+            node.left === schemaMock.document.processingStatus &&
+            node.right === 'processing'
+        )
+    )
+    const processingFreshness = flattenMockConditions(processingState).find(
+      (node: MockCondition) =>
+        node.type === 'gte' && node.left === schemaMock.document.processingStartedAt
+    )
+    expect(processingFreshness?.right).toEqual(
+      new Date(now.getTime() - DOCUMENT_PROCESSING_STALE_THRESHOLD_MS)
+    )
+    expect(
+      dbChainMockFns.set.mock.calls.some(
+        (call) =>
+          (call[0] as Record<string, unknown> | undefined)?.processingQueuedAt === null &&
+          'processingAttempts' in ((call[0] as Record<string, unknown> | undefined) ?? {})
+      )
+    ).toBe(false)
+  })
+
   it('uses the direct fallback outside a run when the deployment flag is off', async () => {
     setEnvFlags({ isTriggerDevEnabled: false })
     Object.assign(env, { TRIGGER_SECRET_KEY: 'trigger-secret' })