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

* fix(knowledge): leave a connector whose credential was removed unscheduled instead of walking the failure ladder to disabled

* fix(knowledge): report a partial connector sync as an outcome instead of failing the run

* fix(knowledge): count a connector's documents before the completion locks and index the live set

* fix(billing): serve execution admission from the usage gate cache instead of re-summing the billing period per run

* fix(knowledge): share the unscheduled connector write, scope credential removal to content-engine modes, and keep both sync holds visible

* fix(knowledge): keep the corpus size on a failed completion count, unschedule only connectors left without an auth source, and give fixture connectors a credential

* fix(knowledge): skip a required-key connector without a key and write the unscheduled state only over the row the run observed

* fix(knowledge): release a dispatch-marked pending row when its credential is missing

* fix(knowledge): assert the billing owner before the credential-missing terminal write

* fix(knowledge): stop offering reconnect once a connector holds a credential again

Author: waleedlatif1

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 {