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

Author: waleedlatif1

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,7 +65,6 @@ export async function executeConnectorSyncJob(payload: unknown) {
       processingDispatchFailed: result.processingDispatch.failed,
     })
 
-    const outcome = classifyConnectorSyncResult(result)
     if (outcome === 'failed') {
       /**
        * `executeSync` has already persisted its terminal state, and retrying this
@@ -85,10 +74,12 @@ export async function executeConnectorSyncJob(payload: unknown) {
        * 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, 'failed'))
+      throw new AbortTaskRunError(`Connector sync failed for ${connectorId}: ${result.error}`)
     }
-    if (outcome === 'partial') {
-      logger.warn(`[${requestId}] ${formatConnectorSyncFailure(connectorId, result, 'partial')}`)
+    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`
+      )
     }
 
     return {

apps/sim/connectors/auth.ts

@@ -34,9 +34,11 @@ export function connectorHasAuthSource(
   auth: ConnectorAuthConfig,
   connector: { credentialId: string | null; encryptedApiKey: string | null }
 ): boolean {
-  if (auth.mode === 'apiKey') return true
-  if (connector.credentialId) return true
-  return Boolean(getConnectorApiKeyConfig(auth) && connector.encryptedApiKey)
+  return (
+    auth.mode === 'apiKey' ||
+    Boolean(connector.credentialId) ||
+    Boolean(getConnectorApiKeyConfig(auth) && connector.encryptedApiKey)
+  )
 }
 
 /** Workspace token input supported by a connector, independent of its member OAuth method. */

apps/sim/lib/credentials/deletion.test.ts

@@ -204,22 +204,29 @@ describe('clearCredentialRefs', () => {
     const updates = capturedQueries.filter((query) =>
       query.sql.startsWith('update "knowledge_connector"')
     )
-    expect(updates).toHaveLength(1)
+    expect(updates).toHaveLength(2)
     const statement = normalizeSql(updates[0].sql)
     expect(statement).toContain('"credential_id" = $')
+    expect(statement).toContain('"last_sync_error" = $')
     expect(statement).toContain('"next_sync_at" = $')
     expect(statement).toContain('"sync_lock_token" = $')
     expect(statement).toContain('"sync_lock_lease_at" = $')
     expect(statement).toContain(
       `CASE WHEN "knowledge_connector"."status" IN ('paused', 'disabled')`
     )
+    expect(statement).toContain('"access_mode" in ($')
     expect(updates[0].params).toEqual(
       expect.arrayContaining([
-        null,
         'Credential removed. Reconnect the connector to resume syncing.',
         'credential-target',
+        'workspace',
+        'admin',
       ])
     )
+    const rest = normalizeSql(updates[1].sql)
+    expect(rest).toContain('"credential_id" = $')
+    expect(rest).not.toContain('"last_sync_error"')
+    expect(updates[1].params).toEqual(expect.arrayContaining([null, 'credential-target']))
   })
 
   it('propagates database failures', async () => {

apps/sim/lib/credentials/deletion.ts

@@ -2,7 +2,7 @@ import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
 import * as schema from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq, notExists, or, sql } from 'drizzle-orm'
+import { and, eq, inArray, notExists, or, sql } from 'drizzle-orm'
 import type { AnyPgColumn, PgTable } from 'drizzle-orm/pg-core'
 import type { NextRequest } from 'next/server'
 import {
@@ -11,7 +11,9 @@ import {
   resourceScopeFromOwner,
 } from '@/lib/core/resource-scope'
 import { resourceScopeCondition } from '@/lib/core/resource-scope.server'
+import { CONTENT_ENGINE_ACCESS_MODES } from '@/lib/knowledge/connectors/access-modes'
 import { CREDENTIAL_REMOVED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+import { buildSyncUnscheduledUpdate } from '@/lib/knowledge/connectors/sync-lock'
 import { CREDENTIAL_SUBBLOCK_IDS } from '@/lib/workflows/persistence/utils'
 
 const logger = createLogger('CredentialDeletion')
@@ -338,24 +340,29 @@ async function readWorkspaceCredentialRefs(
 }
 
 /**
- * A connector whose credential is gone cannot sync until it is reconnected, so the scheduler
- * must stop dispatching it: `nextSyncAt` leaves the due sweep, the status and error name the
- * cause for the reconnect prompt, and the lock is released so a live run's terminal write cannot
- * resurrect a schedule (same transition as a deleted knowledge base). Paused and disabled
- * connectors keep their status.
+ * A content-engine connector whose credential is gone cannot sync until it is reconnected, so
+ * it leaves the due sweep with the reconnect error; paused and disabled connectors keep their
+ * status. A members-mode connector only loses its optional dedicated content credential: its
+ * member crawls keep running, so it merely drops the reference.
  */
 async function clearInKnowledgeConnectors(credentialId: string): Promise<void> {
+  const now = new Date()
   await db
     .update(schema.knowledgeConnector)
     .set({
+      ...buildSyncUnscheduledUpdate(now, CREDENTIAL_REMOVED_SYNC_ERROR),
       credentialId: null,
       status: sql`CASE WHEN ${schema.knowledgeConnector.status} IN ('paused', 'disabled') THEN ${schema.knowledgeConnector.status} ELSE 'error' END`,
-      lastSyncError: CREDENTIAL_REMOVED_SYNC_ERROR,
-      nextSyncAt: null,
-      syncLockToken: null,
-      syncLockLeaseAt: null,
-      updatedAt: new Date(),
     })
+    .where(
+      and(
+        eq(schema.knowledgeConnector.credentialId, credentialId),
+        inArray(schema.knowledgeConnector.accessMode, [...CONTENT_ENGINE_ACCESS_MODES])
+      )
+    )
+  await db
+    .update(schema.knowledgeConnector)
+    .set({ credentialId: null, updatedAt: now })
     .where(eq(schema.knowledgeConnector.credentialId, credentialId))
 }
 

apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts

@@ -525,6 +525,33 @@ describe('content pass checkpoint intent', () => {
     expect(pass.checkpoint.permissionFailures).toBe(true)
   })
 
+  it('reports an incomplete listing alongside unverified permissions', async () => {
+    sourceBody = { value: '<p>Current content</p>' }
+    const checkpoint = {
+      ...beginListingCheckpoint({
+        fingerprint: 'a'.repeat(64),
+        generationId: 'prior',
+        startedAt: new Date(0),
+      }),
+      listingFailures: {
+        count: 1,
+        samples: [
+          {
+            scope: 'unavailable@example.com',
+            operation: 'gmail.threads.list',
+            status: 400,
+            reasons: ['failedPrecondition'],
+          },
+        ],
+      },
+    }
+    mocks.onPage.mockResolvedValue({ permissionsIncomplete: true })
+    const { pass } = await runPass({ checkpoint, access: 'admin' })
+    const lines = pass.holdNotice?.split('\n') ?? []
+    expect(lines).toContain(SOURCE_PERMISSION_ERROR)
+    expect(lines.some((line) => line.includes('unlisted documents were kept'))).toBe(true)
+  })
+
   it('clears permission failure evidence for a newly verified crawl', async () => {
     mocks.onPage.mockResolvedValue({ permissionsIncomplete: false })
     const { pass } = await runPass({ access: 'admin' })

apps/sim/lib/knowledge/connectors/sync-content-pass.ts

@@ -330,12 +330,18 @@ export async function runConnectorContentPass(input: ContentPassInput) {
   const reconciliation = checkpoint.complete
     ? await reconcileCompletedListing(input, checkpoint, withLease)
     : { finished: false, notice: null }
+  /** Unverified permissions and an incomplete listing are independent holds; an admin needs to see both, one per line. */
+  const holdNotice =
+    [
+      checkpoint.permissionFailures ? SOURCE_PERMISSION_ERROR : null,
+      reconciliation.notice ?? (checkpoint.contentFailures ? SOURCE_CONTENT_ERROR : null),
+    ]
+      .filter(Boolean)
+      .join('\n') || null
   return {
     checkpoint,
     complete: checkpoint.complete && reconciliation.finished,
-    holdNotice: checkpoint.permissionFailures
-      ? SOURCE_PERMISSION_ERROR
-      : (reconciliation.notice ?? (checkpoint.contentFailures ? SOURCE_CONTENT_ERROR : null)),
+    holdNotice,
     hydratedCount,
   }
 }

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

@@ -63,6 +63,7 @@ import {
 import {
   assertSyncLeaseHeldInTx,
   buildSyncLockAcquisition,
+  buildSyncUnscheduledUpdate,
   createContentSyncLease,
   holdsSyncLockToken,
   LOCKABLE_CONNECTOR_STATUSES,
@@ -352,6 +353,22 @@ export function isContentPassIncomplete(
   )
 }
 
+/** Live documents the connector owns: what the connector list shows as its document count. */
+async function countLiveConnectorDocuments(connectorId: string): Promise<number> {
+  const [row] = await db
+    .select({ count: sql<number>`count(*)::int` })
+    .from(document)
+    .where(
+      and(
+        eq(document.connectorId, connectorId),
+        eq(document.userExcluded, false),
+        isNull(document.archivedAt),
+        isNull(document.deletedAt)
+      )
+    )
+  return row?.count ?? 0
+}
+
 /**
  * Atomically publishes the completed log and connector terminal state.
  *
@@ -394,9 +411,11 @@ export async function completeSuccessfulSync(
   const completionNotice =
     [directoryNotice, listingNotice, contentNotice].filter(Boolean).join('\n') || null
   /**
-   * A display snapshot, taken before the completion transaction so a slow count neither holds
-   * the connector lock nor turns a sync whose documents already landed into a failure. When it
-   * cannot be read the previous count stands.
+   * Read before the completion transaction so a slow count neither holds the connector lock
+   * nor turns a sync whose documents already landed into a failure. It is also the previous
+   * owned count the next pass's mass-deletion guard starts from, so a tally of this run's
+   * adds and deletes would drift; when it cannot be read the previous count stands, which the
+   * guard already treats as a floor.
    */
   const actualDocCount = await countLiveConnectorDocuments(connectorId).catch((error: unknown) => {
     logger.warn('Could not count connector documents; keeping the previous count', {
@@ -569,14 +588,7 @@ async function releaseSyncLockOnDeletedConnector(
 ): Promise<void> {
   await db
     .update(knowledgeConnector)
-    .set({
-      status: 'error',
-      nextSyncAt: null,
-      lastSyncError: 'Connector deleted during sync',
-      syncLockToken: null,
-      syncLockLeaseAt: null,
-      updatedAt: new Date(),
-    })
+    .set(buildSyncUnscheduledUpdate(new Date(), 'Connector deleted during sync'))
     .where(holdsSyncLockToken(connectorId, syncLogId))
 }
 
@@ -684,13 +696,8 @@ export function buildSyncCapacityUpdate(
   errorMessage: string
 ) {
   return {
-    status: 'error' as const,
-    lastSyncError: errorMessage,
-    nextSyncAt: null,
+    ...buildSyncUnscheduledUpdate(now, errorMessage),
     consecutiveFailures: previousFailures ?? 0,
-    syncLockToken: null,
-    syncLockLeaseAt: null,
-    updatedAt: now,
   }
 }
 
@@ -703,21 +710,6 @@ export function buildSyncCapacityUpdate(
  * `consecutiveFailures` still resets: a held pass is a healthy sync that declined
  * to delete, not a failure, and marking it broken would stop it syncing at all.
  */
-/** Live documents the connector owns: what the connector list shows as its document count. */
-async function countLiveConnectorDocuments(connectorId: string): Promise<number> {
-  const [row] = await db
-    .select({ count: sql<number>`count(*)::int` })
-    .from(document)
-    .where(
-      and(
-        eq(document.connectorId, connectorId),
-        eq(document.userExcluded, false),
-        isNull(document.archivedAt),
-        isNull(document.deletedAt)
-      )
-    )
-  return row?.count ?? 0
-}
 
 export function buildSyncSuccessUpdate(
   now: Date,
@@ -850,21 +842,13 @@ export async function executeSync(
   /**
    * A connector with no token source cannot succeed, and each attempt would only walk the
    * failure ladder and, at its end, disable a connector that merely needs reconnecting. Left
-   * unscheduled with the reconnect error instead; the same terminal transition as a deleted
-   * knowledge base, so a live run's own terminal write cannot revive the schedule.
+   * unscheduled with the reconnect error instead, the same transition as a deleted knowledge base.
    */
   if (!connectorHasAuthSource(connectorConfig.auth, connectorBeforeLock)) {
     logger.warn('Skipping sync: connector has no credential to authenticate with', { connectorId })
     await db
       .update(knowledgeConnector)
-      .set({
-        status: 'error',
-        nextSyncAt: null,
-        lastSyncError: CREDENTIAL_REMOVED_SYNC_ERROR,
-        syncLockToken: null,
-        syncLockLeaseAt: null,
-        updatedAt: new Date(),
-      })
+      .set(buildSyncUnscheduledUpdate(new Date(), CREDENTIAL_REMOVED_SYNC_ERROR))
       .where(eq(knowledgeConnector.id, connectorId))
     return { ...result, skipReason: 'credential_missing' }
   }
@@ -890,25 +874,7 @@ export async function executeSync(
     )
     await db
       .update(knowledgeConnector)
-      .set({
-        status: 'error',
-        nextSyncAt: null,
-        lastSyncError: 'Knowledge base deleted',
-        /**
-         * Clears the lock alongside the status.
-         *
-         * This write runs BEFORE the lock is taken, but it is unconditional on
-         * status, so it can land on a row a previous run left `syncing` — a run
-         * that may still be alive. Flipping status without releasing the token
-         * left a row that was neither locked nor reclaimable: the reaper only
-         * looks at `syncing` rows, and the old run's terminal write could still
-         * match its own token and resurrect a state for a knowledge base that no
-         * longer exists. Releasing both makes the transition terminal.
-         */
-        syncLockToken: null,
-        syncLockLeaseAt: null,
-        updatedAt: new Date(),
-      })
+      .set(buildSyncUnscheduledUpdate(new Date(), 'Knowledge base deleted'))
       .where(eq(knowledgeConnector.id, connectorId))
     return { ...result, skipReason: 'knowledge_base_deleted' }
   }

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

@@ -69,6 +69,25 @@ export function holdsSyncLockToken(connectorId: string, syncLockToken: string) {
 /** Connector statuses a scheduler may start an automatic run from. */
 export const RUNNABLE_CONNECTOR_STATUSES = ['active', 'error'] as const
 
+/**
+ * A terminal, unscheduled error. The lock is released alongside the status because
+ * this write can land on a row a previous run left `syncing` — a run that may still
+ * be alive. Flipping status without releasing the token left a row that was neither
+ * locked nor reclaimable: the reaper only looks at `syncing` rows, and the old run's
+ * terminal write could still match its own token and resurrect the schedule.
+ * Releasing both makes the transition terminal.
+ */
+export function buildSyncUnscheduledUpdate(now: Date, errorMessage: string) {
+  return {
+    status: 'error' as const,
+    lastSyncError: errorMessage,
+    nextSyncAt: null,
+    syncLockToken: null,
+    syncLockLeaseAt: null,
+    updatedAt: now,
+  }
+}
+
 /**
  * The statuses a run may take the lock from.
  *