fix(knowledge): unschedule a connector whose credential the source rejected and prompt to reconnect

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connector-recovery.tsx

@@ -4,7 +4,10 @@ import { useEffect, useState } from 'react'
 import { Chip } from '@sim/emcn'
 import type { ConnectorData } from '@/lib/api/contracts/knowledge/connectors'
 import { type ResourceScope, resourceScopeFields } from '@/lib/core/resource-scope'
-import { CREDENTIAL_REMOVED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+import {
+  CREDENTIAL_REMOVED_SYNC_ERROR,
+  CREDENTIAL_REVOKED_SYNC_ERROR,
+} from '@/lib/knowledge/connectors/sync-limits'
 import { getCanonicalScopesForProvider, getProviderIdFromServiceId } from '@/lib/oauth'
 import { getMissingRequiredScopes } from '@/lib/oauth/utils'
 import { ConnectOAuthModal } from '@/app/workspace/[workspaceId]/components/connect-oauth-modal'
@@ -82,7 +85,10 @@ export function ConnectorRecovery({
   const docsUrl = isSearchIndex ? connectorDef?.searchDocsUrl : undefined
   const credentialRemoved =
     connector.lastSyncError === CREDENTIAL_REMOVED_SYNC_ERROR && !connector.credentialId
-  const pausedTitle = credentialRemoved
+  const credentialRevoked =
+    connector.lastSyncError === CREDENTIAL_REVOKED_SYNC_ERROR && Boolean(connector.credentialId)
+  const reconnectRequired = credentialRemoved || credentialRevoked
+  const pausedTitle = reconnectRequired
     ? 'Reconnect to resume syncing'
     : 'Sync paused after repeated failures'
 
@@ -97,7 +103,7 @@ export function ConnectorRecovery({
           }
         />
       )}
-      {connector.status === 'disabled' || credentialRemoved ? (
+      {connector.status === 'disabled' || reconnectRequired ? (
         <SettingsResourceRow
           title={
             !canEdit

apps/sim/connectors/types.ts

@@ -266,6 +266,7 @@ export const SYNC_SKIP_REASONS = [
   'sync_superseded',
   'connector_deleted_during_sync',
   'credential_missing',
+  'credential_revoked',
 ] as const
 
 export type SyncSkipReason = (typeof SYNC_SKIP_REASONS)[number]

apps/sim/lib/credentials/draft-hooks.test.ts

@@ -115,6 +115,15 @@ describe('handleReconnectCredential', () => {
     expect(mocks.clearDeadFlag).toHaveBeenCalledWith(
       getOAuthRefreshCoordinationIdentity('account-new')
     )
+    /** Connectors the rejected credential had unscheduled are due again. */
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({
+        status: 'active',
+        lastSyncError: null,
+        consecutiveFailures: 0,
+        nextSyncAt: new Date('2026-08-14T18:00:00.000Z'),
+      })
+    )
     expect(auditMockFns.mockRecordAudit).toHaveBeenCalledWith(
       expect.objectContaining({
         resourceId: 'credential-1',

apps/sim/lib/credentials/draft-hooks.ts

@@ -6,6 +6,7 @@ import { getPostgresConstraintName, getPostgresErrorCode } from '@sim/utils/erro
 import { generateId } from '@sim/utils/id'
 import { and, eq, sql } from 'drizzle-orm'
 import { deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
+import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
 import { clearOAuthRefreshDeadFlag } from '@/lib/oauth/refresh-coordination'
 import { captureServerEvent } from '@/lib/posthog/server'
 
@@ -75,6 +76,7 @@ export async function handleCreateCredentialFromDraft(params: {
       .where(eq(schema.credential.id, existingCredential.id))
 
     await clearOAuthRefreshDeadFlag(accountId)
+    await resumeConnectorsAfterCredentialReconnect(existingCredential.id, now)
 
     recordAudit({
       workspaceId: draft.workspaceId,
@@ -209,6 +211,7 @@ export async function handleReconnectCredential(params: {
   )
 
   await clearOAuthRefreshDeadFlag(newAccountId)
+  await resumeConnectorsAfterCredentialReconnect(draft.credentialId, now)
 
   recordAudit({
     workspaceId,

apps/sim/lib/credentials/organization-draft.ts

@@ -8,6 +8,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { resourceScopeCondition } from '@/lib/core/resource-scope.server'
 import { deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
 import { getCredentialCreationOrganizationContext } from '@/lib/credentials/organization'
+import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
 import { clearOAuthRefreshDeadFlag } from '@/lib/oauth/refresh-coordination'
 
 /** Completes the exact draft bound to the authenticated provider callback, rechecking current ownership under membership locks. */
@@ -124,6 +125,7 @@ export async function completeOrganizationCredentialDraft(input: {
     }
   })
   await clearOAuthRefreshDeadFlag(input.accountId)
+  if (result.reconnected) await resumeConnectorsAfterCredentialReconnect(result.credentialId, now)
   recordAudit({
     actorId: input.userId,
     action: result.reconnected

apps/sim/lib/knowledge/connectors/credential-recovery.test.ts

@@ -0,0 +1,31 @@
+/**
+ * @vitest-environment node
+ */
+import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
+import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+
+describe('resumeConnectorsAfterCredentialReconnect', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('puts only the connectors the rejected credential unscheduled back on schedule', async () => {
+    const now = new Date('2026-09-22T20:00:00.000Z')
+    await resumeConnectorsAfterCredentialReconnect('credential-1', now)
+    expect(dbChainMockFns.set).toHaveBeenCalledWith({
+      status: 'active',
+      lastSyncError: null,
+      consecutiveFailures: 0,
+      nextSyncAt: now,
+      updatedAt: now,
+    })
+    const guard = JSON.stringify(dbChainMockFns.where.mock.calls.at(-1))
+    expect(guard).toContain('knowledgeConnector.credentialId')
+    expect(guard).toContain('credential-1')
+    expect(guard).toContain('knowledgeConnector.status')
+    expect(guard).toContain(CREDENTIAL_REVOKED_SYNC_ERROR)
+  })
+})

apps/sim/lib/knowledge/connectors/credential-recovery.ts

@@ -0,0 +1,36 @@
+import { db } from '@sim/db'
+import { knowledgeConnector } from '@sim/db/schema'
+import { and, eq } from 'drizzle-orm'
+import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+
+/**
+ * Puts the connectors a reconnected credential had unscheduled back on their schedule.
+ *
+ * A sync that finds its credential rejected by the source leaves the connector unscheduled
+ * with {@link CREDENTIAL_REVOKED_SYNC_ERROR}, since retrying cannot help until someone
+ * authorizes again. Reauthorizing the same credential is that moment: every connector still
+ * carrying that error is due now, with its failure count cleared. Connectors that were paused
+ * or disabled for another reason keep their state, and a connector that already moved on is
+ * left alone.
+ */
+export async function resumeConnectorsAfterCredentialReconnect(
+  credentialId: string,
+  now: Date
+): Promise<void> {
+  await db
+    .update(knowledgeConnector)
+    .set({
+      status: 'active',
+      lastSyncError: null,
+      consecutiveFailures: 0,
+      nextSyncAt: now,
+      updatedAt: now,
+    })
+    .where(
+      and(
+        eq(knowledgeConnector.credentialId, credentialId),
+        eq(knowledgeConnector.status, 'error'),
+        eq(knowledgeConnector.lastSyncError, CREDENTIAL_REVOKED_SYNC_ERROR)
+      )
+    )
+}

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

@@ -3,6 +3,7 @@
  */
 import {
   authOAuthUtilsMock,
+  authOAuthUtilsMockFns,
   dbChainMockFns,
   drizzleOrmMock,
   flattenMockConditions,
@@ -17,6 +18,7 @@ import { DrizzleQueryError } from 'drizzle-orm/errors'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import * as connectorTokens from '@/lib/knowledge/connectors/access-token'
 import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine'
+import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
 import {
   classifySuspectListing,
   evaluateListingSafety,
@@ -3007,6 +3009,84 @@ describe('executeSync heartbeats during the listing phase', () => {
     }
   )
 
+  /** A locked OAuth connector whose token resolution the test controls. */
+  function primeOAuthRunUpToToken() {
+    const oauthConnector = {
+      ...CONNECTOR,
+      connectorType: 'oauth',
+      credentialId: 'cred-1',
+      accessMode: 'workspace',
+    }
+    queueTableRows(schemaMock.knowledgeConnector, [oauthConnector])
+    for (let i = 0; i < 20; i++)
+      queueTableRows(schemaMock.knowledgeConnector, [
+        { id: 'c-1', connectorArchivedAt: null, connectorDeletedAt: null, kbDeletedAt: null },
+      ])
+    queueTableRows(schemaMock.knowledgeBase, [{ userId: 'u-1', workspaceId: 'ws-1' }])
+    dbChainMockFns.returning.mockReset()
+    dbChainMockFns.returning.mockResolvedValueOnce([oauthConnector])
+    /** The terminal write lands on the row this run still holds. */
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
+    const tokenUser = vi
+      .spyOn(connectorTokens, 'resolveConnectorTokenUserId')
+      .mockResolvedValueOnce('u-1')
+    const resolveToken = vi
+      .spyOn(connectorTokens, 'resolveConnectorAccessToken')
+      .mockResolvedValueOnce(null)
+    return () => {
+      tokenUser.mockRestore()
+      resolveToken.mockRestore()
+    }
+  }
+
+  it('unschedules a connector whose credential the source rejected instead of retrying it', async () => {
+    const restore = primeOAuthRunUpToToken()
+    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError.mockResolvedValueOnce(
+      'invalid_grant'
+    )
+    try {
+      const result = await executeSync('c-1', {
+        billingAttribution: { workspaceId: 'ws-1' } as never,
+      })
+      expect(result.skipReason).toBe('credential_revoked')
+      expect(result.error).toBeUndefined()
+      expect(authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError).toHaveBeenCalledWith(
+        'cred-1'
+      )
+      expect(dbChainMockFns.set).toHaveBeenCalledWith(
+        expect.objectContaining({
+          status: 'error',
+          nextSyncAt: null,
+          lastSyncError: CREDENTIAL_REVOKED_SYNC_ERROR,
+          syncLockToken: null,
+          syncLockLeaseAt: null,
+        })
+      )
+      expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+        expect.objectContaining({ consecutiveFailures: expect.any(Number) })
+      )
+    } finally {
+      restore()
+    }
+  })
+
+  it('keeps the failure ladder for a credential that resolved no token without a terminal error', async () => {
+    const restore = primeOAuthRunUpToToken()
+    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError.mockResolvedValueOnce(null)
+    try {
+      const result = await executeSync('c-1', {
+        billingAttribution: { workspaceId: 'ws-1' } as never,
+      })
+      expect(result.skipReason).toBeUndefined()
+      expect(result.error).toContain('Failed to obtain access token')
+      expect(dbChainMockFns.set).toHaveBeenCalledWith(
+        expect.objectContaining({ status: 'error', consecutiveFailures: 1 })
+      )
+    } finally {
+      restore()
+    }
+  })
+
   it.each([
     { acl: undefined, incomplete: true },
     { acl: ['invalid-token'], incomplete: true },

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

@@ -57,6 +57,7 @@ import {
   CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES,
   CONNECTOR_SYNC_MAX_DURATION_SECONDS,
   CREDENTIAL_REMOVED_SYNC_ERROR,
+  CREDENTIAL_REVOKED_SYNC_ERROR,
   connectorFailureBackoffMinutes,
   MAX_CONSECUTIVE_FAILURES,
 } from '@/lib/knowledge/connectors/sync-limits'
@@ -88,6 +89,7 @@ import {
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import { getRetryAfterMs, isRateLimitError } from '@/lib/knowledge/documents/utils'
 import { ensureSourceVectorIndex } from '@/lib/knowledge/search/source-vector-indexes'
+import { getCredentialTerminalRefreshError } from '@/lib/oauth/credential-service'
 import { connectorHasAuthSource } from '@/connectors/auth'
 import { CONNECTOR_REGISTRY } from '@/connectors/registry.server'
 import type {
@@ -743,9 +745,26 @@ export function buildSyncSuccessUpdate(
   }
 }
 
+/**
+ * A credential the source rejected outright: the refresh path recorded a terminal error for
+ * it, so no retry can produce a token until the credential is reauthorized.
+ */
+export class ConnectorCredentialRevokedError extends Error {
+  constructor(
+    readonly credentialId: string,
+    readonly errorCode: string
+  ) {
+    super(`Credential ${credentialId} was rejected by the source (${errorCode})`)
+    this.name = 'ConnectorCredentialRevokedError'
+  }
+}
+
 /**
  * Resolves the token a connector syncs with, failing loudly where the shared
  * resolver reports "no token" — a sync has no reconnect prompt to fall back to.
+ * A credential the source has rejected outright fails as
+ * {@link ConnectorCredentialRevokedError}, so the run can unschedule the
+ * connector instead of walking the failure ladder toward a retry that cannot help.
  */
 async function resolveAccessToken(
   connector: { credentialId: string | null; encryptedApiKey: string | null },
@@ -770,6 +789,13 @@ async function resolveAccessToken(
       userId,
       authMode: connectorConfig.auth.mode,
     })
+    const terminalError =
+      connectorConfig.auth.mode === 'oauth' && connector.credentialId
+        ? await getCredentialTerminalRefreshError(connector.credentialId)
+        : null
+    if (terminalError && connector.credentialId) {
+      throw new ConnectorCredentialRevokedError(connector.credentialId, terminalError)
+    }
     throw new Error(`Failed to obtain access token for credential ${connector.credentialId}`)
   }
 
@@ -1427,6 +1453,46 @@ export async function executeSync(
         }
       }
 
+      if (error instanceof ConnectorCredentialRevokedError) {
+        /**
+         * Retrying cannot help until the credential is reauthorized, so the
+         * connector leaves its schedule with a reconnect prompt instead of
+         * climbing the failure ladder toward the same rejection. Reauthorizing
+         * the credential puts it back on schedule. The run itself is a skip:
+         * nothing about the source failed, and a sync that cannot start is not
+         * an incident to page on.
+         */
+        logger.warn('Sync unscheduled: the source rejected the connector credential', {
+          connectorId,
+          credentialId: error.credentialId,
+          errorCode: error.errorCode,
+        })
+        try {
+          await completeSyncLog(syncLogId, 'failed', result, {
+            errorMessage: CREDENTIAL_REVOKED_SYNC_ERROR,
+          })
+          const landed = await writeTerminalConnectorState(
+            connectorId,
+            syncLogId,
+            buildSyncUnscheduledUpdate(new Date(), CREDENTIAL_REVOKED_SYNC_ERROR)
+          )
+          if (!landed) {
+            logger.warn(
+              'Unschedule discarded — connector was reclaimed while this run was executing',
+              { connectorId, syncLogId }
+            )
+          }
+        } catch (recoveryError) {
+          logger.error('Failed to unschedule the connector', {
+            connectorId,
+            error:
+              getConnectorFailureDiagnostic(recoveryError)?.message ??
+              toError(recoveryError).message,
+          })
+        }
+        return { ...result, skipReason: 'credential_revoked' }
+      }
+
       const diagnostic = getConnectorFailureDiagnostic(error)
       const errorMessage = diagnostic?.message ?? toError(error).message
       const retryAfterMs = getRetryAfterMs(error)

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

@@ -25,6 +25,13 @@ export const MAX_CONSECUTIVE_FAILURES = 10
 export const CREDENTIAL_REMOVED_SYNC_ERROR =
   'Credential removed. Reconnect the connector to resume syncing.'
 
+/**
+ * The error a connector carries once the source rejects its credential outright (a revoked or
+ * expired grant, not a passing failure); cleared by reauthorizing that credential.
+ */
+export const CREDENTIAL_REVOKED_SYNC_ERROR =
+  'The source no longer accepts this credential. Reconnect it to resume syncing.'
+
 /**
  * The error a connector carries once {@link MAX_CONSECUTIVE_FAILURES} disables it.
  *