fix(knowledge): resume by reconnected account across a Slack installation, recheck the rejection before unscheduling, and fail a run that cannot record it

Author: waleedlatif1

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

@@ -103,6 +103,7 @@ describe('handleReconnectCredential', () => {
       { id: 'credential-1', accountId: null, displayName: 'Renamed Gmail' },
     ])
     queueTableRows(schemaMock.credential, [])
+    queueTableRows(schemaMock.account, [{ providerId: 'gmail', accountId: 'subject-new' }])
 
     await handleReconnectCredential({
       draft: { credentialId: 'credential-1' },

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

@@ -76,7 +76,7 @@ export async function handleCreateCredentialFromDraft(params: {
       .where(eq(schema.credential.id, existingCredential.id))
 
     await clearOAuthRefreshDeadFlag(accountId)
-    await resumeConnectorsAfterCredentialReconnect(existingCredential.id, now)
+    await resumeConnectorsAfterCredentialReconnect(accountId, now)
 
     recordAudit({
       workspaceId: draft.workspaceId,
@@ -211,7 +211,7 @@ export async function handleReconnectCredential(params: {
   )
 
   await clearOAuthRefreshDeadFlag(newAccountId)
-  await resumeConnectorsAfterCredentialReconnect(draft.credentialId, now)
+  await resumeConnectorsAfterCredentialReconnect(newAccountId, now)
 
   recordAudit({
     workspaceId,

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

@@ -125,7 +125,7 @@ export async function completeOrganizationCredentialDraft(input: {
     }
   })
   await clearOAuthRefreshDeadFlag(input.accountId)
-  if (result.reconnected) await resumeConnectorsAfterCredentialReconnect(result.credentialId, now)
+  if (result.reconnected) await resumeConnectorsAfterCredentialReconnect(input.accountId, now)
   recordAudit({
     actorId: input.userId,
     action: result.reconnected

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

@@ -1,31 +1,52 @@
 /**
  * @vitest-environment node
  */
-import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } 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'
 
+const RESUMED = {
+  status: 'active',
+  lastSyncError: null,
+  consecutiveFailures: 0,
+}
+
 describe('resumeConnectorsAfterCredentialReconnect', () => {
+  const now = new Date('2026-09-22T20:00:00.000Z')
+
   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,
-    })
+  it('resumes the connectors of every credential on the reconnected account', async () => {
+    queueTableRows(schemaMock.account, [
+      { providerId: 'confluence', providerAccountId: 'subject-1' },
+    ])
+    await resumeConnectorsAfterCredentialReconnect('account-1', now)
+    expect(dbChainMockFns.set).toHaveBeenCalledWith({ ...RESUMED, 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)
+    const credentials = JSON.stringify(dbChainMockFns.where.mock.calls)
+    expect(credentials).toContain('"left":"credential.accountId","right":"account-1"')
+  })
+
+  it('resumes across the Slack installation, whose sibling accounts share the repaired chain', async () => {
+    queueTableRows(schemaMock.account, [
+      { providerId: 'slack', providerAccountId: 'TEXAMPLE-usr_U1' },
+    ])
+    await resumeConnectorsAfterCredentialReconnect('account-1', now)
+    expect(dbChainMockFns.set).toHaveBeenCalledWith({ ...RESUMED, nextSyncAt: now, updatedAt: now })
+    const conditions = JSON.stringify(dbChainMockFns.where.mock.calls)
+    expect(conditions).toContain('"pattern":"TEXAMPLE-%"')
+    expect(conditions).not.toContain('"left":"credential.accountId","right":"account-1"')
+  })
+
+  it('does nothing for an account that no longer exists', async () => {
+    queueTableRows(schemaMock.account, [])
+    await resumeConnectorsAfterCredentialReconnect('account-gone', now)
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
   })
 })

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

@@ -1,22 +1,40 @@
 import { db } from '@sim/db'
-import { knowledgeConnector } from '@sim/db/schema'
-import { and, eq } from 'drizzle-orm'
+import { account, credential, knowledgeConnector } from '@sim/db/schema'
+import { and, eq, inArray } from 'drizzle-orm'
 import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+import { extractSlackTeamId, installationFilter, isSlackProvider } from '@/lib/oauth/slack'
 
 /**
- * Puts the connectors a reconnected credential had unscheduled back on their schedule.
+ * Puts the connectors a reconnected account 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.
+ * authorizes again. Reauthorizing the account is that moment: every connector on a credential
+ * of that account still carrying the error is due now, with its failure count cleared. A Slack
+ * reauthorization repairs the installation's shared token chain, so the connectors on every
+ * credential of the installation's sibling accounts are due as well. Connectors 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,
+  accountId: string,
   now: Date
 ): Promise<void> {
+  const [reconnected] = await db
+    .select({ providerId: account.providerId, providerAccountId: account.accountId })
+    .from(account)
+    .where(eq(account.id, accountId))
+    .limit(1)
+  if (!reconnected) return
+  const slackTeamId = isSlackProvider(reconnected.providerId)
+    ? extractSlackTeamId(reconnected.providerAccountId)
+    : null
+  const repairedAccounts = slackTeamId
+    ? inArray(
+        credential.accountId,
+        db.select({ id: account.id }).from(account).where(installationFilter(slackTeamId))
+      )
+    : eq(credential.accountId, accountId)
   await db
     .update(knowledgeConnector)
     .set({
@@ -28,7 +46,10 @@ export async function resumeConnectorsAfterCredentialReconnect(
     })
     .where(
       and(
-        eq(knowledgeConnector.credentialId, credentialId),
+        inArray(
+          knowledgeConnector.credentialId,
+          db.select({ id: credential.id }).from(credential).where(repairedAccounts)
+        ),
         eq(knowledgeConnector.status, 'error'),
         eq(knowledgeConnector.lastSyncError, CREDENTIAL_REVOKED_SYNC_ERROR)
       )

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

@@ -3041,9 +3041,10 @@ describe('executeSync heartbeats during the listing phase', () => {
 
   it('unschedules a connector whose credential the source rejected instead of retrying it', async () => {
     const restore = primeOAuthRunUpToToken()
-    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError.mockResolvedValueOnce(
-      'invalid_grant'
-    )
+    /** Rejected at token resolution and still rejected when the run records its outcome. */
+    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
+      .mockResolvedValueOnce('invalid_grant')
+      .mockResolvedValueOnce('invalid_grant')
     try {
       const result = await executeSync('c-1', {
         billingAttribution: { workspaceId: 'ws-1' } as never,
@@ -3070,6 +3071,52 @@ describe('executeSync heartbeats during the listing phase', () => {
     }
   })
 
+  it('takes the failure ladder when the credential was reauthorized while the run was failing', async () => {
+    const restore = primeOAuthRunUpToToken()
+    /** Rejected at token resolution, repaired by the time the run records its outcome. */
+    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
+      .mockResolvedValueOnce('invalid_grant')
+      .mockResolvedValueOnce(null)
+    try {
+      const result = await executeSync('c-1', {
+        billingAttribution: { workspaceId: 'ws-1' } as never,
+      })
+      expect(result.skipReason).toBeUndefined()
+      expect(result.error).toContain('rejected by the source')
+      expect(dbChainMockFns.set).toHaveBeenCalledWith(
+        expect.objectContaining({ status: 'error', consecutiveFailures: 1 })
+      )
+      expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+        expect.objectContaining({ lastSyncError: CREDENTIAL_REVOKED_SYNC_ERROR })
+      )
+    } finally {
+      restore()
+    }
+  })
+
+  it('reports a run that could not record the unschedule as failed, not skipped', async () => {
+    const restore = primeOAuthRunUpToToken()
+    /** Rejected at token resolution and still rejected when the run records its outcome. */
+    authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
+      .mockResolvedValueOnce('invalid_grant')
+      .mockResolvedValueOnce('invalid_grant')
+    /** The terminal write fails after the lock CAS consumed the first result. */
+    dbChainMockFns.returning.mockReset()
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      { ...CONNECTOR, connectorType: 'oauth', credentialId: 'cred-1', accessMode: 'workspace' },
+    ])
+    dbChainMockFns.returning.mockRejectedValueOnce(new Error('connection reset'))
+    try {
+      const result = await executeSync('c-1', {
+        billingAttribution: { workspaceId: 'ws-1' } as never,
+      })
+      expect(result.skipReason).toBeUndefined()
+      expect(result.error).toContain('connection reset')
+    } 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)

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

@@ -1458,39 +1458,54 @@ export async function executeSync(
          * 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.
+         * the credential puts it back on schedule, and a reauthorization that
+         * landed while this run was failing has already cleared the rejection:
+         * that run takes the ordinary ladder below, so its next attempt uses the
+         * repaired chain rather than leaving a repaired connector unscheduled.
+         * The unscheduled run itself is a skip: nothing about the source failed,
+         * and a sync that cannot start is not an incident to page on. A run that
+         * cannot record the unschedule is a failure, so the runner reports it
+         * instead of leaving the connector locked behind a benign outcome.
          */
-        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(
+        const stillRejected = await getCredentialTerminalRefreshError(error.credentialId)
+        if (stillRejected) {
+          logger.warn('Sync unscheduled: the source rejected the connector credential', {
             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 }
+            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)
             )
-          }
-        } catch (recoveryError) {
-          logger.error('Failed to unschedule the connector', {
-            connectorId,
-            error:
+            if (!landed) {
+              logger.warn(
+                'Unschedule discarded — connector was reclaimed while this run was executing',
+                { connectorId, syncLogId }
+              )
+            }
+            return { ...result, skipReason: 'credential_revoked' }
+          } catch (recoveryError) {
+            const recoveryMessage =
               getConnectorFailureDiagnostic(recoveryError)?.message ??
-              toError(recoveryError).message,
-          })
+              toError(recoveryError).message
+            logger.error('Failed to unschedule the connector', {
+              connectorId,
+              error: recoveryMessage,
+            })
+            result.error = recoveryMessage
+            return result
+          }
         }
-        return { ...result, skipReason: 'credential_revoked' }
+        logger.info('Credential reauthorized during the run; the retry uses the repaired chain', {
+          connectorId,
+          credentialId: error.credentialId,
+        })
       }
 
       const diagnostic = getConnectorFailureDiagnostic(error)

apps/sim/lib/oauth/slack.ts

@@ -30,7 +30,8 @@ export function extractSlackTeamId(externalAccountId: string | null | undefined)
   return match ? match[1] : null
 }
 
-function installationFilter(teamId: string) {
+/** The account rows of one Slack installation: every bot token row of the team. */
+export function installationFilter(teamId: string) {
   return and(eq(account.providerId, 'slack'), like(account.accountId, `${teamId}-%`))
 }