fix(knowledge): preserve live document processing during recovery (#7993)

* fix(knowledge): preserve live document processing during recovery

* fix(knowledge): recover abandoned redelivery and cancel liveness requests

* fix(knowledge): fence recovery against concurrent protection

Author: waleedlatif1

apps/sim/lib/internal/mistral/client.test.ts

@@ -3,6 +3,11 @@
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
+const { errorLog } = vi.hoisted(() => ({ errorLog: vi.fn() }))
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({ error: errorLog, info: vi.fn(), warn: vi.fn(), debug: vi.fn() }),
+}))
+
 const { fetchPinned, admit, settle, validate } = vi.hoisted(() => ({
   fetchPinned: vi.fn(),
   admit: vi.fn(),
@@ -63,9 +68,41 @@ describe('Mistral provider transport', () => {
     expect(settle).toHaveBeenCalledWith('success', undefined)
   })
 
+  it('records cooldown without waiting for a stalled 429 error body', async () => {
+    const cancel = vi.fn()
+    fetchPinned.mockResolvedValue(
+      new Response(new ReadableStream({ cancel }), {
+        status: 429,
+        headers: { 'retry-after': '60' },
+      })
+    )
+    await expect(submitMistralOcr('private-key', {})).rejects.toMatchObject({
+      reason: 'rate_limit',
+      retryAfterMs: 60_000,
+    })
+    expect(settle).toHaveBeenCalledWith('rate_limit', 60_000)
+    expect(cancel).toHaveBeenCalledOnce()
+    expect(fetchPinned).toHaveBeenCalledOnce()
+  })
+
+  it('preserves provider rejection when its diagnostic body exceeds the byte limit', async () => {
+    fetchPinned.mockResolvedValue(new Response('x'.repeat(70_000), { status: 400 }))
+    await expect(submitMistralOcr('private-key', {})).rejects.toMatchObject({
+      status: 400,
+      body: { success: false, error: 'Mistral API error: HTTP 400' },
+    })
+    expect(errorLog).toHaveBeenCalledWith(
+      'Mistral API error',
+      expect.objectContaining({ status: 400, bodyFormat: 'unavailable' })
+    )
+  })
+
   it('identifies provider request rejection without retaining echoed document contents', async () => {
     fetchPinned.mockResolvedValue(
-      Response.json({ message: 'Sensitive fixture document text' }, { status: 400 })
+      Response.json(
+        { type: 'invalid_request_error', code: 400, message: 'Sensitive fixture document text' },
+        { status: 400, headers: { 'x-request-id': 'ocr-request-123' } }
+      )
     )
     await expect(submitMistralOcr('key', {})).rejects.toMatchObject({
       source: 'provider',
@@ -74,6 +111,16 @@ describe('Mistral provider transport', () => {
     })
     expect(fetchPinned).toHaveBeenCalledOnce()
     expect(settle).toHaveBeenCalledWith('failure', undefined)
+    expect(errorLog).toHaveBeenCalledWith(
+      'Mistral API error',
+      expect.objectContaining({
+        status: 400,
+        providerRequestId: 'ocr-request-123',
+        providerErrorCode: '400',
+        providerErrorType: 'invalid_request_error',
+      })
+    )
+    expect(JSON.stringify(errorLog.mock.calls)).not.toContain('Sensitive fixture document text')
   })
 
   it.each([

apps/sim/lib/internal/mistral/client.ts

@@ -11,9 +11,10 @@ import {
   validateUrlWithDNS,
 } from '@/lib/core/security/input-validation.server'
 import { getMistralCapacityConfig, getMistralCapacityScope } from '@/lib/internal/mistral/capacity'
+import { getOcrResponseDiagnostic } from '@/lib/internal/mistral/error-diagnostics'
 import { MistralOperationError } from '@/lib/internal/mistral/errors'
 import { MISTRAL_OCR_REQUEST_POLICY } from '@/lib/knowledge/documents/ocr-request-policy'
-import { readBoundedHttpErrorBody, resolveRetryDelayMs } from '@/lib/knowledge/documents/utils'
+import { readBoundedHttpErrorPayload, resolveRetryDelayMs } from '@/lib/knowledge/documents/utils'
 
 const logger = createLogger('MistralClient')
 const MISTRAL_ENDPOINT = 'https://api.mistral.ai/v1/ocr'
@@ -138,8 +139,13 @@ export async function submitMistralOcr(
             retryAfterMs,
           })
         }
-        await readBoundedHttpErrorBody(response)
-        logger.error('Mistral API error', { status: response.status })
+        const payload = await readBoundedHttpErrorPayload(response)
+        logger.error('Mistral API error', {
+          provider: 'mistral',
+          operation: 'ocr',
+          status: response.status,
+          ...getOcrResponseDiagnostic(response.headers, payload.ok ? payload.body : ''),
+        })
         throw new MistralOperationError(
           response.status,
           { success: false, error: `Mistral API error: HTTP ${response.status}` },

apps/sim/lib/internal/mistral/error-diagnostics.test.ts

@@ -0,0 +1,49 @@
+/** @vitest-environment node */
+import { describe, expect, it } from 'vitest'
+import { getOcrResponseDiagnostic } from '@/lib/internal/mistral/error-diagnostics'
+
+describe('OCR error diagnostics', () => {
+  it.each([
+    { code: 400, type: 'invalid_request_error', message: 'private document' },
+    { error: { code: 400, type: 'invalid_request_error', message: 'private document' } },
+  ])('projects only safe fields from a provider envelope', (body) => {
+    expect(
+      getOcrResponseDiagnostic(new Headers({ 'x-request-id': 'request_123' }), JSON.stringify(body))
+    ).toEqual({
+      bodyFormat: 'json',
+      providerRequestId: 'request_123',
+      providerErrorCode: '400',
+      providerErrorType: 'invalid_request_error',
+    })
+  })
+
+  it.each(['not json', '<html>private input</html>', '', 'null', '[]', '42'])(
+    'tolerates an unavailable or non-object error: %s',
+    (body) => {
+      expect(getOcrResponseDiagnostic(new Headers(), body)).toMatchObject({
+        providerRequestId: null,
+        providerErrorCode: null,
+        providerErrorType: null,
+      })
+    }
+  )
+
+  it('does not trust arbitrary codes, error types, echoed input, or request IDs', () => {
+    const diagnostic = getOcrResponseDiagnostic(
+      new Headers({ 'x-request-id': 'secret '.repeat(30), 'apim-request-id': 'safe-123' }),
+      JSON.stringify({
+        code: 'private_document',
+        type: { input: 'private_document' },
+        message: 'private_document',
+        param: 'secret-key',
+        detail: { input: 'private_document' },
+      })
+    )
+    expect(diagnostic).toMatchObject({
+      providerRequestId: 'safe-123',
+      providerErrorCode: 'unrecognized',
+      providerErrorType: 'unrecognized',
+    })
+    expect(JSON.stringify(diagnostic)).not.toMatch(/private_document|secret-key/)
+  })
+})

apps/sim/lib/internal/mistral/error-diagnostics.ts

@@ -0,0 +1,57 @@
+const SAFE_ERROR_CODES = new Set([
+  'invalid_request_error',
+  'authentication_error',
+  'permission_error',
+  'rate_limit_error',
+  'server_error',
+  'unknown_model',
+  'BadRequest',
+  'InvalidRequest',
+  'DeploymentNotFound',
+  'ResourceNotFound',
+  'OperationNotSupported',
+  'Unauthorized',
+  'Forbidden',
+  'TooManyRequests',
+  'InternalServerError',
+  'ServiceUnavailable',
+])
+
+function safeCode(value: unknown): string | null {
+  if (value === undefined || value === null) return null
+  if (typeof value === 'string' && SAFE_ERROR_CODES.has(value)) return value
+  if (typeof value === 'number' && Number.isInteger(value) && value >= 400 && value <= 599)
+    return String(value)
+  return 'unrecognized'
+}
+
+function safeRequestId(value: string | null): string | null {
+  return value && /^[a-zA-Z0-9_-]{1,128}$/.test(value) ? value : null
+}
+
+/** Projects bounded OCR error responses without retaining messages, document data or URLs. */
+export function getOcrResponseDiagnostic(headers: Pick<Headers, 'get'>, body: string) {
+  const diagnostic = {
+    providerRequestId:
+      safeRequestId(headers.get('x-request-id')) ??
+      safeRequestId(headers.get('apim-request-id')) ??
+      safeRequestId(headers.get('x-ms-request-id')),
+    bodyFormat: body ? 'non_json' : 'unavailable',
+    providerErrorCode: null as string | null,
+    providerErrorType: null as string | null,
+  }
+  if (!body) return diagnostic
+  let parsed: unknown
+  try {
+    parsed = JSON.parse(body)
+  } catch {
+    return diagnostic
+  }
+  diagnostic.bodyFormat = 'json'
+  if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return diagnostic
+  const error = 'error' in parsed ? parsed.error : parsed
+  if (!error || typeof error !== 'object' || Array.isArray(error)) return diagnostic
+  diagnostic.providerErrorCode = safeCode('code' in error ? error.code : undefined)
+  diagnostic.providerErrorType = safeCode('type' in error ? error.type : undefined)
+  return diagnostic
+}