fix(connectors): surface a held reconciliation and scope the cap to eligible rows

Review round 1 on #6909.

- A held pass reported an ordinary successful sync: the cap logged an error and
  the success update then cleared lastSyncError and reset consecutiveFailures,
  so source-removed documents stayed indexed with no operator signal. The notice
  is now threaded into the success update itself — writing it at the hold site
  would have been clobbered by that same update ~300 lines later, in the same
  run. status stays active and the failure counter still resets: a held pass is
  a healthy sync that declined to delete.
- The cap denominator counted excluded tombstones, which partitionSyncReconciliation
  can never delete, inflating a budget against rows that cannot be spent. Both
  sides are now counted over the deletion-eligible population, matching the
  numerator.
- completeSyncLog now only writes a row still marked started, so a late-finishing
  in-process run cannot overwrite a row the stale sweep already closed. The TTL
  is documented as a hard ceiling for both dispatch paths.

Author: waleedlatif1

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

@@ -1,7 +1,15 @@
 /**
  * @vitest-environment node
  */
-import { authOAuthUtilsMock } from '@sim/testing'
+import {
+  authOAuthUtilsMock,
+  dbChainMockFns,
+  drizzleOrmMock,
+  hasMockCondition,
+  type MockCondition,
+  resetDbChainMock,
+  schemaMock,
+} from '@sim/testing'
 import { generateShortId } from '@sim/utils/id'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import {
@@ -12,13 +20,7 @@ import {
 } from '@/lib/knowledge/connectors/sync-engine'
 import type { ExternalDocument } from '@/connectors/types'
 
-vi.mock('drizzle-orm', () => ({
-  and: vi.fn(),
-  eq: vi.fn(),
-  inArray: vi.fn(),
-  isNull: vi.fn(),
-  ne: vi.fn(),
-}))
+vi.mock('drizzle-orm', () => drizzleOrmMock)
 vi.mock('@/lib/knowledge/documents/service', () => ({
   hardDeleteDocuments: vi.fn(),
   isTriggerAvailable: vi.fn(),
@@ -961,3 +963,131 @@ describe('countNonExcludedListed', () => {
     expect(classifySuspectListing(listed.size, ownedDocCount)).toBeNull()
   })
 })
+
+describe('countDeletionEligibleOwned', () => {
+  const doc = (id: string) => ({ id, externalId: id })
+  const excluded = (id: string) => ({ id, externalId: id, userExcluded: true })
+
+  it('does not let excluded tombstones inflate the denominator', async () => {
+    const { countDeletionEligibleOwned } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(countDeletionEligibleOwned([doc('a')], [excluded('t1'), excluded('t2')])).toBe(1)
+    expect(countDeletionEligibleOwned([doc('a')], [doc('t1'), excluded('t2')])).toBe(2)
+  })
+
+  it('excludes user-excluded rows from the live side too', async () => {
+    const { countDeletionEligibleOwned } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(countDeletionEligibleOwned([doc('a'), excluded('b')], [])).toBe(1)
+  })
+
+  it('agrees with the numerator on which population it counts', async () => {
+    const { classifySuspectListing, countDeletionEligibleOwned, countNonExcludedListed } =
+      await import('@/lib/knowledge/connectors/sync-engine')
+
+    /**
+     * 100 live + 100 excluded tombstones. Counting the excluded tombstones would
+     * put the denominator at 200 and hide a listing that returned nothing but
+     * excluded documents.
+     */
+    const existing = Array.from({ length: 100 }, (_, i) => doc(`live-${i}`))
+    const tombstoned = Array.from({ length: 100 }, (_, i) => excluded(`ex-${i}`))
+    const listed = new Set(tombstoned.map((d) => d.externalId))
+    const excludedExternalIds = new Set(listed)
+
+    const ownedDocCount = countDeletionEligibleOwned(existing, tombstoned)
+    const listedDocCount = countNonExcludedListed(listed, excludedExternalIds)
+
+    expect(ownedDocCount).toBe(100)
+    expect(listedDocCount).toBe(0)
+    expect(classifySuspectListing(listedDocCount, ownedDocCount)).toBe('empty')
+  })
+})
+
+describe('buildReconciliationHoldNotice', () => {
+  it('names the counts and the full-sync remedy', async () => {
+    const { buildReconciliationHoldNotice } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const notice = buildReconciliationHoldNotice(500, 250, 1000)
+
+    expect(notice).toContain('500')
+    expect(notice).toContain('250')
+    expect(notice).toContain('1000')
+    expect(notice).toContain('full sync')
+  })
+})
+
+describe('buildSyncSuccessUpdate', () => {
+  const now = new Date('2026-08-20T00:00:00.000Z')
+
+  it('carries a hold notice into lastSyncError instead of clearing it', async () => {
+    const { buildSyncSuccessUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    /**
+     * The sequencing assertion. This update runs at the end of the sync, long
+     * after the hold is detected, so writing the notice at the hold site would
+     * be clobbered here.
+     */
+    const update = buildSyncSuccessUpdate(now, 42, null, 'held: 500 removals withheld')
+
+    expect(update.lastSyncError).toBe('held: 500 removals withheld')
+  })
+
+  it('still clears lastSyncError on an ordinary successful sync', async () => {
+    const { buildSyncSuccessUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(buildSyncSuccessUpdate(now, 42, null, null).lastSyncError).toBeNull()
+  })
+
+  it('does not treat a held pass as a broken connector', async () => {
+    const { buildSyncSuccessUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const update = buildSyncSuccessUpdate(now, 42, null, 'held')
+
+    expect(update.status).toBe('active')
+    expect(update.consecutiveFailures).toBe(0)
+  })
+})
+
+describe('completeSyncLog', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('only writes a row that is still started', async () => {
+    const { completeSyncLog } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    await completeSyncLog('log-1', 'completed', {
+      docsAdded: 1,
+      docsUpdated: 0,
+      docsDeleted: 0,
+      docsUnchanged: 0,
+      docsFailed: 0,
+    })
+
+    const where = dbChainMockFns.where.mock.calls[0][0]
+    /**
+     * Without this the sweep and a late-finishing in-process run race: the sweep
+     * marks the row failed, then the run overwrites it as completed.
+     */
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnectorSyncLog.status &&
+          node.right === 'started'
+      )
+    ).toBe(true)
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnectorSyncLog.id &&
+          node.right === 'log-1'
+      )
+    ).toBe(true)
+  })
+})

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

@@ -295,7 +295,20 @@ function calculateNextSyncTime(syncIntervalMinutes: number): Date | null {
   return new Date(now + syncIntervalMinutes * 60_000 + jitterMs)
 }
 
-async function completeSyncLog(
+/**
+ * Records a sync run's outcome on its log row.
+ *
+ * Guarded on `status = 'started'` so a run that outlives
+ * {@link CONNECTOR_SYNC_STALE_LOCK_TTL_MS} cannot overwrite a row the
+ * scheduler's stale sweep already closed. Without the guard the two writers
+ * race and produce contradictory history: the sweep marks the row `failed`,
+ * then the still-running sync reports `completed` on the same row.
+ *
+ * A no-op on the normal path — nothing else touches the row between its
+ * `started` insert and this call, so the guard only ever bites once the sweep
+ * has declared the run dead, and the sweep's verdict is the one that stands.
+ */
+export async function completeSyncLog(
   syncLogId: string,
   status: 'completed' | 'failed',
   result: SyncResult,
@@ -313,7 +326,12 @@ async function completeSyncLog(
       docsUnchanged: result.docsUnchanged,
       docsFailed: result.docsFailed,
     })
-    .where(eq(knowledgeConnectorSyncLog.id, syncLogId))
+    .where(
+      and(
+        eq(knowledgeConnectorSyncLog.id, syncLogId),
+        eq(knowledgeConnectorSyncLog.status, 'started')
+      )
+    )
 }
 
 /**
@@ -458,6 +476,73 @@ export function evaluateListingSafety(
   return { reason, blocked: !corroborated, corroborated }
 }
 
+/**
+ * Documents a reconciliation pass could actually remove.
+ *
+ * Both reads are filtered, not just the tombstoned one: the live read already
+ * excludes `userExcluded` rows in SQL, so filtering it again is a no-op today,
+ * but it keeps this count self-consistent with
+ * {@link partitionSyncReconciliation}, which gates deletion on the same flag for
+ * both lists. The result is the denominator for the deletion cap and for
+ * {@link classifySuspectListing}, whose numerator
+ * ({@link countNonExcludedListed}) ranges over the same population.
+ */
+export function countDeletionEligibleOwned(
+  existingDocs: ReconciliationDoc[],
+  tombstonedDocs: ReconciliationDoc[]
+): number {
+  return (
+    existingDocs.filter((d) => !d.userExcluded).length +
+    tombstonedDocs.filter((d) => !d.userExcluded).length
+  )
+}
+
+/**
+ * Operator-facing explanation of a held reconciliation pass.
+ *
+ * Stored on `knowledgeConnector.lastSyncError` because a hold is otherwise
+ * invisible: the sync completes normally and an operator sees an ordinary green
+ * run while source-removed documents stay indexed. Names the forced full sync,
+ * which is the documented way to apply the removals once the source is verified.
+ */
+export function buildReconciliationHoldNotice(
+  requested: number,
+  cap: number,
+  ownedDocCount: number
+): string {
+  return (
+    `Withheld ${requested} document removal(s) — more than the ${cap} allowed in one sync ` +
+    `of ${ownedDocCount} documents. Documents deleted at the source are still indexed. ` +
+    'Check the source is returning its full contents, then run a full sync to apply the removals.'
+  )
+}
+
+/**
+ * The connector row a successful sync writes.
+ *
+ * `holdNotice` is threaded through rather than written when the hold is detected
+ * because this update runs at the very end of the sync and would otherwise clear
+ * `lastSyncError` in the same run. `status` stays `active` and
+ * `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.
+ */
+export function buildSyncSuccessUpdate(
+  now: Date,
+  actualDocCount: number,
+  nextSyncAt: Date | null,
+  holdNotice: string | null
+) {
+  return {
+    status: 'active' as const,
+    lastSyncAt: now,
+    lastSyncError: holdNotice,
+    lastSyncDocCount: actualDocCount,
+    nextSyncAt,
+    consecutiveFailures: 0,
+    updatedAt: now,
+  }
+}
+
 /**
  * The document count to attribute to the previous sync when reconstructing its
  * listing.
@@ -1391,7 +1476,14 @@ export async function executeSync(
      * same thing. Only evaluated when reconciliation would otherwise run, so
      * healthy syncs pay nothing and no existing gate is loosened.
      */
-    const ownedDocCount = existingDocs.length + tombstonedDocs.length
+    /**
+     * Counted over deletion-eligible rows on both sides. The live read filters
+     * excluded documents in SQL; the tombstoned read only projects the flag, so
+     * excluded tombstones must be dropped here or they inflate a denominator
+     * governing a population they are not part of. Matches `listedDocCount`,
+     * which `countNonExcludedListed` already puts on the same footing.
+     */
+    const ownedDocCount = countDeletionEligibleOwned(existingDocs, tombstonedDocs)
     /**
      * Counted over the same population as `ownedDocCount`: excluded documents
      * are absent from the live read, so they must not inflate the numerator.
@@ -1441,7 +1533,23 @@ export async function executeSync(
       ownedDocCount,
       options?.fullSync
     )
+    /**
+     * Surfaced on the connector so a held pass is visible to an operator rather
+     * than only in logs: without it the sync completes green, clears
+     * `lastSyncError`, and source-removed documents stay indexed with no signal.
+     * Written through the success update at the end of this run rather than
+     * here — that update sets `lastSyncError: null` unconditionally and would
+     * otherwise clobber this within the same sync. `status` is deliberately left
+     * `active`: the sync itself succeeded, and marking the connector broken
+     * would stop it syncing at all.
+     */
+    let reconciliationHoldNotice: string | null = null
     if (capped.held) {
+      reconciliationHoldNotice = buildReconciliationHoldNotice(
+        capped.requested,
+        capped.cap,
+        ownedDocCount
+      )
       logger.error('Reconciliation deletions held — exceeds per-sync blast-radius cap', {
         connectorId,
         connectorType: connector.connectorType,
@@ -1685,15 +1793,14 @@ export async function executeSync(
     const now = new Date()
     await db
       .update(knowledgeConnector)
-      .set({
-        status: 'active',
-        lastSyncAt: now,
-        lastSyncError: null,
-        lastSyncDocCount: actualDocCount,
-        nextSyncAt: calculateNextSyncTime(connector.syncIntervalMinutes),
-        consecutiveFailures: 0,
-        updatedAt: now,
-      })
+      .set(
+        buildSyncSuccessUpdate(
+          now,
+          actualDocCount,
+          calculateNextSyncTime(connector.syncIntervalMinutes),
+          reconciliationHoldNotice
+        )
+      )
       .where(
         and(
           eq(knowledgeConnector.id, connectorId),

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

@@ -12,6 +12,21 @@ export const CONNECTOR_SYNC_MAX_DURATION_SECONDS = 3600
  * MUST stay above {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS}: reclaiming frees the
  * lock for another sync, so a TTL at or below the run ceiling would start a second
  * sync while the first is still writing, both racing the same documents.
+ *
+ * This is a hard ceiling for BOTH execution paths, not just the queued one. A
+ * Trigger.dev run is killed at {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS}, so it
+ * is provably dead well before this. The fallback path is not: when Trigger.dev is
+ * unavailable, `dispatchSync` runs `executeSync` fire-and-forget inside the web
+ * process with no duration cap, and such a run genuinely can still be executing
+ * when this TTL expires.
+ *
+ * Treating it as dead anyway is deliberate. An unbounded background sync in a
+ * recyclable web process that has run for two hours is indistinguishable from one
+ * whose process was recycled out from under it, and the cost of guessing wrong in
+ * the other direction is a connector locked out of syncing forever. The sweep's
+ * verdict is therefore authoritative: `completeSyncLog` is guarded on
+ * `status = 'started'`, so a late finisher cannot overwrite a row already closed
+ * here, and it loses the race by design rather than by accident.
  */
 export const CONNECTOR_SYNC_STALE_LOCK_TTL_MS = CONNECTOR_SYNC_MAX_DURATION_SECONDS * 2 * 1000