fix(knowledge): make connector syncs resilient to removed credentials, partial outcomes and slow completion counts

apps/sim/app/api/v1/knowledge/search/route.test.ts

@@ -70,7 +70,9 @@ vi.mock('@/lib/billing/calculations/usage-monitor', () => ({
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   resolveBillingAttribution: mockResolveBillingAttribution,
   resolveSystemBillingAttribution: mockResolveSystemBillingAttribution,
-  checkAttributedUsageLimits: vi.fn().mockResolvedValue({ isExceeded: false }),
+}))
+vi.mock('@/lib/billing/core/usage-gate-cache', () => ({
+  checkSearchUsageLimits: vi.fn().mockResolvedValue({ isExceeded: false }),
 }))
 
 vi.mock('@/lib/knowledge/embeddings', () => ({

apps/sim/app/api/v1/knowledge/search/route.ts

@@ -2,10 +2,10 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { v1KnowledgeSearchContract } from '@/lib/api/contracts/v1/knowledge'
 import { parseRequest } from '@/lib/api/server'
 import {
-  checkAttributedUsageLimits,
   resolveBillingAttribution,
   resolveSystemBillingAttribution,
 } from '@/lib/billing/core/billing-attribution'
+import { checkSearchUsageLimits } from '@/lib/billing/core/usage-gate-cache'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { ALL_TAG_SLOTS } from '@/lib/knowledge/constants'
 import { toKbEmbeddingDimensions } from '@/lib/knowledge/embedding-models'
@@ -82,7 +82,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
      * keys resolve their system actor and immutable payer from one workspace read.
      */
     if (billingAttribution) {
-      const usage = await checkAttributedUsageLimits(billingAttribution)
+      const usage = await checkSearchUsageLimits(billingAttribution)
       if (usage.isExceeded) {
         return NextResponse.json(
           { error: usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.' },

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

@@ -4,6 +4,7 @@ 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 { getCanonicalScopesForProvider, getProviderIdFromServiceId } from '@/lib/oauth'
 import { getMissingRequiredScopes } from '@/lib/oauth/utils'
 import { ConnectOAuthModal } from '@/app/workspace/[workspaceId]/components/connect-oauth-modal'
@@ -79,6 +80,11 @@ export function ConnectorRecovery({
   }
 
   const docsUrl = isSearchIndex ? connectorDef?.searchDocsUrl : undefined
+  const credentialRemoved =
+    connector.lastSyncError === CREDENTIAL_REMOVED_SYNC_ERROR && !connector.credentialId
+  const pausedTitle = credentialRemoved
+    ? 'Reconnect to resume syncing'
+    : 'Sync paused after repeated failures'
 
   return (
     <>
@@ -91,16 +97,16 @@ export function ConnectorRecovery({
           }
         />
       )}
-      {connector.status === 'disabled' ? (
+      {connector.status === 'disabled' || credentialRemoved ? (
         <SettingsResourceRow
           title={
             !canEdit
-              ? 'Sync paused after repeated failures'
+              ? pausedTitle
               : requiresAccountSettings
                 ? 'Update the source account, then resume syncing'
                 : serviceId
                   ? 'Reconnect to resume syncing'
-                  : 'Sync paused after repeated failures'
+                  : pausedTitle
           }
           trailing={
             canEdit && requiresAccountSettings && onEdit ? (

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

@@ -8,6 +8,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { SyncLogData } from '@/lib/api/contracts/knowledge/connectors'
 import {
   CONNECTOR_SYNC_STALE_LOCK_TTL_MS,
+  CREDENTIAL_REMOVED_SYNC_ERROR,
   MEMBER_SYNC_STALE_LOCK_TTL_MS,
 } from '@/lib/knowledge/connectors/sync-limits'
 
@@ -480,6 +481,21 @@ describe('Connector credential reauthorization', () => {
     expect(container.textContent).toContain('No connected sources yet.')
   })
 
+  it('offers reconnect for a connector whose credential was removed', () => {
+    oauthCredentialsState.current = [
+      { id: 'credential-1', name: 'Workspace Slack', provider: 'slack-custom' },
+    ]
+    const container = renderSection(
+      makeConnector({
+        status: 'error',
+        credentialId: null,
+        lastSyncError: CREDENTIAL_REMOVED_SYNC_ERROR,
+      })
+    )
+    expect(container.textContent).toContain('Reconnect to resume syncing')
+    expect(findButton(container, 'Reconnect')).not.toBeDisabled()
+  })
+
   it('reauthorizes with the resolved credential provider and identity', () => {
     oauthCredentialsState.current = [
       { id: 'credential-1', name: 'Workspace Slack', provider: 'slack-custom' },

apps/sim/background/drain-governed-subject.test.ts

@@ -49,9 +49,11 @@ vi.mock('@/enrichments/run', () => ({
 }))
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
-  checkAttributedUsageLimits: mocks.checkAttributedUsageLimits,
   toBillingContext: () => ({}),
 }))
+vi.mock('@/lib/billing/core/usage-gate-cache', () => ({
+  checkExecutionUsageLimits: mocks.checkAttributedUsageLimits,
+}))
 vi.mock('@/lib/table/rows/secret-provenance', () => ({
   createExactEmptyTableRowSecretProvenance: () => ({ complete: true, columns: {} }),
   createTableRowSecretProvenanceFromRegistry: () => ({ complete: true, columns: {} }),

apps/sim/background/enrichment-capability-subject.test.ts

@@ -58,9 +58,11 @@ vi.mock('@/enrichments/run', () => ({
 }))
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   assertBillingAttributionSnapshot: vi.fn((value) => value),
-  checkAttributedUsageLimits: mocks.checkAttributedUsageLimits,
   toBillingContext: vi.fn(() => ({})),
 }))
+vi.mock('@/lib/billing/core/usage-gate-cache', () => ({
+  checkExecutionUsageLimits: mocks.checkAttributedUsageLimits,
+}))
 vi.mock('@/lib/table/rows/secret-provenance', () => ({
   createExactEmptyTableRowSecretProvenance: vi.fn(() => undefined),
   createTableRowSecretProvenanceFromRegistry: vi.fn(() => undefined),

apps/sim/background/knowledge-connector-sync.test.ts

@@ -5,12 +5,16 @@
 import { AbortTaskRunError } from '@trigger.dev/sdk'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockAssertConnectorSyncPayload, mockExecuteSync, mockTask } = vi.hoisted(() => ({
+const { mockAssertConnectorSyncPayload, mockExecuteSync, mockTask, mockWarn } = vi.hoisted(() => ({
+  mockWarn: vi.fn(),
   mockAssertConnectorSyncPayload: vi.fn(),
   mockExecuteSync: vi.fn(),
   mockTask: vi.fn((config) => config),
 }))
 
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({ info: vi.fn(), warn: mockWarn, error: vi.fn(), debug: vi.fn() }),
+}))
 vi.mock('@trigger.dev/sdk', () => ({
   task: mockTask,
   AbortTaskRunError: class AbortTaskRunError extends Error {},
@@ -114,7 +118,7 @@ describe('knowledge connector sync worker', () => {
     })
   })
 
-  it('fails visibly without retrying an already-persisted partial sync', async () => {
+  it('returns a partial sync as an outcome instead of failing the run', async () => {
     mockAssertConnectorSyncPayload.mockReturnValue({
       connectorId: 'connector-1',
       requestId: 'request-1',
@@ -136,8 +140,15 @@ describe('knowledge connector sync worker', () => {
       billingAttribution: BILLING_ATTRIBUTION,
     })
 
-    await expect(run).rejects.toBeInstanceOf(AbortTaskRunError)
-    await expect(run).rejects.toThrow('Connector sync partially failed')
+    await expect(run).resolves.toMatchObject({
+      success: false,
+      outcome: 'partial',
+      docsFailed: 1,
+      processingDispatch: { failed: 1 },
+    })
+    expect(mockWarn).toHaveBeenCalledWith(
+      expect.stringContaining('1 source failures, 1 dispatch failures')
+    )
   })
 
   it('completes a durably scheduled capacity wait while preserving existing source failures', async () => {
@@ -163,7 +174,7 @@ describe('knowledge connector sync worker', () => {
       deferred: waiting.deferred,
     })
     mockExecuteSync.mockResolvedValue({ ...waiting, docsFailed: 1 })
-    await expect(executeConnectorSyncJob({})).rejects.toThrow('partially failed')
+    expect(await executeConnectorSyncJob({})).toMatchObject({ outcome: 'partial', success: false })
     mockExecuteSync.mockResolvedValue({ ...waiting, error: 'Retry persistence failed' })
     await expect(executeConnectorSyncJob({})).rejects.toThrow('Retry persistence failed')
   })
@@ -266,6 +277,7 @@ describe('knowledge connector sync worker', () => {
       outcome: 'partial',
       listingIncomplete: true,
     })
+    expect(mockWarn).not.toHaveBeenCalled()
   })
 
   it('classifies a persisted connector error as a failed task', () => {

apps/sim/background/knowledge-connector-sync.ts

@@ -27,17 +27,6 @@ export function classifyConnectorSyncResult(result: SyncResult): ConnectorSyncTa
   return 'completed'
 }
 
-function formatConnectorSyncFailure(
-  connectorId: string,
-  result: SyncResult,
-  outcome: Extract<ConnectorSyncTaskOutcome, 'partial' | 'failed'>
-): string {
-  if (outcome === 'failed') {
-    return `Connector sync failed for ${connectorId}: ${result.error}`
-  }
-  return `Connector sync partially failed for ${connectorId}: ${result.docsFailed} source failures, ${result.processingDispatch.failed} dispatch failures`
-}
-
 export async function executeConnectorSyncJob(payload: unknown) {
   const {
     connectorId,
@@ -60,9 +49,10 @@ export async function executeConnectorSyncJob(payload: unknown) {
       dispatchToken,
     })
 
+    const outcome = classifyConnectorSyncResult(result)
     logger.info(`[${requestId}] Connector sync completed`, {
       connectorId,
-      outcome: classifyConnectorSyncResult(result),
+      outcome,
       deferred: result.deferred,
       added: result.docsAdded,
       updated: result.docsUpdated,
@@ -75,17 +65,20 @@ export async function executeConnectorSyncJob(payload: unknown) {
       processingDispatchFailed: result.processingDispatch.failed,
     })
 
-    const outcome = classifyConnectorSyncResult(result)
-    if (outcome === 'failed' || result.docsFailed > 0 || result.processingDispatch.failed > 0) {
+    if (outcome === 'failed') {
       /**
-       * `executeSync` has already persisted its terminal state. Source failures
-       * preserve the previous incremental watermark so the next connector pass
-       * replays them; dispatch failures remain eligible for the stuck-document
-       * sweep. Retrying this whole task immediately would duplicate a large
-       * fan-out, so fail visibly without retrying the completed transaction.
+       * `executeSync` has already persisted its terminal state, and retrying this
+       * whole task would duplicate a large fan-out, so fail visibly without
+       * retrying the completed transaction. A partial sync is not a failed run:
+       * its source failures keep the previous incremental watermark so the next
+       * connector pass replays them, its dispatch failures stay eligible for the
+       * stuck-document sweep, and the outcome rides on the return value.
        */
-      throw new AbortTaskRunError(
-        formatConnectorSyncFailure(connectorId, result, outcome === 'failed' ? 'failed' : 'partial')
+      throw new AbortTaskRunError(`Connector sync failed for ${connectorId}: ${result.error}`)
+    }
+    if (outcome === 'partial' && (result.docsFailed > 0 || result.processingDispatch.failed > 0)) {
+      logger.warn(
+        `[${requestId}] Connector sync partially failed for ${connectorId}: ${result.docsFailed} source failures, ${result.processingDispatch.failed} dispatch failures`
       )
     }
 

apps/sim/background/workflow-column-execution.ts

@@ -10,9 +10,9 @@ import { and, eq, isNull, or } from 'drizzle-orm'
 import {
   assertBillingAttributionSnapshot,
   type BillingAttributionSnapshot,
-  checkAttributedUsageLimits,
   toBillingContext,
 } from '@/lib/billing/core/billing-attribution'
+import { checkExecutionUsageLimits } from '@/lib/billing/core/usage-gate-cache'
 import { checkAndBillPayerOverageThreshold } from '@/lib/billing/threshold-billing'
 import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import {
@@ -575,7 +575,7 @@ async function runWorkflowAndWriteTerminal(
          * Gate the exact workspace payer and member cap before hosted-key cost.
          * A denial clears the cell pre-stamp and surfaces the upgrade state.
          */
-        const usage = await checkAttributedUsageLimits(enrichmentBillingAttribution)
+        const usage = await checkExecutionUsageLimits(enrichmentBillingAttribution)
         if (usage.isExceeded) {
           logger.warn(
             `Usage limit reached — halting enrichment (table=${tableId} row=${rowId} group=${groupId})`

apps/sim/background/workflow-group-governed-subject.test.ts

@@ -93,9 +93,11 @@ vi.mock('@/executor/utils/resolved-secret-trace-registry', () => ({
 }))
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
-  checkAttributedUsageLimits: async () => ({ isExceeded: false }),
   toBillingContext: () => ({}),
 }))
+vi.mock('@/lib/billing/core/usage-gate-cache', () => ({
+  checkExecutionUsageLimits: async () => ({ isExceeded: false }),
+}))
 /** Real pacing would sleep jittered backoff against the global db mock. */
 vi.mock('@/lib/core/rate-limiter/rate-limiter', () => ({
   RateLimiter: class {

apps/sim/connectors/auth.test.ts

@@ -2,11 +2,31 @@
  * @vitest-environment node
  */
 import { describe, expect, it } from 'vitest'
-import { isConnectorCredentialTypeAllowed } from '@/connectors/auth'
+import { connectorHasAuthSource, isConnectorCredentialTypeAllowed } from '@/connectors/auth'
 import { confluenceConnectorMeta } from '@/connectors/confluence/meta'
 import { googleDriveConnectorMeta } from '@/connectors/google-drive/meta'
 import { slackConnectorMeta } from '@/connectors/slack/meta'
 
+describe('connectorHasAuthSource', () => {
+  const none = { credentialId: null, encryptedApiKey: null }
+  const keyed = { credentialId: null, encryptedApiKey: 'enc' }
+  const linked = { credentialId: 'cred', encryptedApiKey: null }
+
+  it('mirrors the token resolver for every auth shape', () => {
+    expect(connectorHasAuthSource({ mode: 'apiKey', label: 'Key' }, none)).toBe(false)
+    expect(connectorHasAuthSource({ mode: 'apiKey', label: 'Key' }, keyed)).toBe(true)
+    expect(connectorHasAuthSource({ mode: 'apiKey', label: 'Key', optional: true }, none)).toBe(
+      true
+    )
+    expect(connectorHasAuthSource({ mode: 'oauth', provider: 'slack' }, none)).toBe(false)
+    expect(connectorHasAuthSource({ mode: 'oauth', provider: 'slack' }, linked)).toBe(true)
+    expect(connectorHasAuthSource({ mode: 'oauth', provider: 'slack' }, keyed)).toBe(false)
+    expect(
+      connectorHasAuthSource({ mode: 'oauth', provider: 'github', apiKey: { label: 'PAT' } }, keyed)
+    ).toBe(true)
+  })
+})
+
 describe('connector credential eligibility', () => {
   it.each([confluenceConnectorMeta, googleDriveConnectorMeta])(
     'requires a service account for $name central indexing and preserves member and workspace OAuth',

apps/sim/connectors/auth.ts

@@ -25,6 +25,20 @@ export function isConnectorCredentialTypeAllowed(
   )
 }
 
+/**
+ * Whether a workspace-mode connector row still carries something to authenticate with. A
+ * connector whose credential was removed keeps its documents but has no token source, so a
+ * sync cannot run until it is reconnected.
+ */
+export function connectorHasAuthSource(
+  auth: ConnectorAuthConfig,
+  connector: { credentialId: string | null; encryptedApiKey: string | null }
+): boolean {
+  const apiKeyConfig = getConnectorApiKeyConfig(auth)
+  if (apiKeyConfig && connector.encryptedApiKey) return true
+  return auth.mode === 'apiKey' ? apiKeyConfig?.optional === true : Boolean(connector.credentialId)
+}
+
 /** Workspace token input supported by a connector, independent of its member OAuth method. */
 export function getConnectorApiKeyConfig(
   auth: ConnectorAuthConfig

apps/sim/connectors/types.ts

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

apps/sim/lib/billing/core/usage-gate-cache.test.ts

@@ -12,6 +12,7 @@ vi.mock('@/lib/billing/core/billing-attribution', () => ({
 
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
 import {
+  checkExecutionUsageLimits,
   checkIngestionUsageLimits,
   checkSearchUsageLimits,
   resetUsageGateCache,
@@ -158,3 +159,23 @@ describe('checkSearchUsageLimits', () => {
     expect(mockCheck).toHaveBeenCalledTimes(2)
   })
 })
+
+describe('checkExecutionUsageLimits', () => {
+  beforeEach(() => {
+    resetUsageGateCache()
+    mockCheck.mockReset().mockResolvedValue({ isExceeded: false })
+  })
+
+  it('reuses an admission across workspaces of the same payer', async () => {
+    await checkExecutionUsageLimits(ATTRIBUTION)
+    await checkExecutionUsageLimits({ ...ATTRIBUTION, workspaceId: 'ws-2' })
+    expect(mockCheck).toHaveBeenCalledTimes(1)
+  })
+
+  it('re-reads a refusal', async () => {
+    mockCheck.mockResolvedValue({ isExceeded: true, message: 'over' })
+    await checkExecutionUsageLimits(ATTRIBUTION)
+    await checkExecutionUsageLimits(ATTRIBUTION)
+    expect(mockCheck).toHaveBeenCalledTimes(2)
+  })
+})