fix(knowledge): walk the members-mode reconcile by external id and tombstone explicit removals without a completed listing

The absence reconcile ordered its resumable walk by source_seen_at, which
every member listing rewrites for what it observed. On a connector larger than
one run's page budget the walk chased re-stamped documents and a pass never
ended, so a document that lost its observers behind the cursor was never
revisited. The walk now keys on external id through
doc_connector_external_id_idx, which a document never changes, so a pass
completes within ceil(documents / budget) runs.

Documents a run explicitly unobserved (a complete listing, a change-feed
withdrawal, or a member removal) are now tombstoned even when no member with a
completed listing remains, as the stale-member sweep already does. Removing
the only listed member previously left every document only it observed live
and unobserved indefinitely. The reconcile and the purge stay gated on a
completed listing.

Author: waleedlatif1

apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts

@@ -3,6 +3,7 @@ import { db } from '@sim/db'
 import {
   document,
   knowledgeConnector,
+  knowledgeConnectorMember,
   knowledgeDocumentObservation,
   organization,
   user,
@@ -21,9 +22,11 @@ import {
   recordMemberObservations,
   removeMemberObservationsForDocuments,
 } from '@/lib/knowledge/connectors/member-observations'
+import { resumeMembershipRewrites } from '@/lib/knowledge/connectors/member-sync-engine'
 import { MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN } from '@/lib/knowledge/connectors/sync-limits'
 import {
   assertSyncLeaseHeldInTx,
+  createMemberSyncLease,
   SyncLockLostException,
   stillHoldsMemberSyncLock,
 } from '@/lib/knowledge/connectors/sync-lock'
@@ -143,10 +146,10 @@ describe('member document lifecycle in PostgreSQL', () => {
     const unobserved = Array.from({ length: pageBudget + 20 }, (_, index) =>
       row(`unobserved-${index}`)
     )
-    const lastSeen = sql`'2026-06-01 00:00:00'::timestamp`
+    /** Both sort after every unobserved document, beyond what this run's reconcile reaches. */
     const [lostByThisRun, stillObservedByBob] = [
-      { ...row('lost-by-this-run'), sourceSeenAt: lastSeen },
-      { ...row('still-observed-by-bob'), sourceSeenAt: lastSeen },
+      row('zz-lost-by-this-run'),
+      row('zz-still-observed-by-bob'),
     ]
     await insertRows([...unobserved, lostByThisRun, stillObservedByBob])
     await observe([lostByThisRun.id, stillObservedByBob.id])
@@ -172,7 +175,7 @@ describe('member document lifecycle in PostgreSQL', () => {
     expect(afterFirst.has(lostByThisRun.id)).toBe(true)
     expect(afterFirst.has(stillObservedByBob.id)).toBe(false)
     expect(unobserved.filter(({ id }) => !afterFirst.has(id))).toHaveLength(20)
-    expect(await savedCursor()).toEqual({ seenAt: expect.any(String), id: expect.any(String) })
+    expect(await savedCursor()).toEqual({ externalId: expect.any(String) })
 
     expect(await run()).toEqual({ tombstoned: 20, resurrected: 0, purged: 0, finished: true })
     const afterSecond = await tombstonedIds()
@@ -181,6 +184,101 @@ describe('member document lifecycle in PostgreSQL', () => {
     expect(await savedCursor()).toBeNull()
   })
 
+  it('tombstones what removing the only listed member unobserved, though no completed listing remains', async () => {
+    const [removedMember, otherMember] = members.members
+    await db
+      .update(knowledgeConnectorMember)
+      .set({
+        lastCompleteListingAt: new Date(),
+        listingCheckpoint: { kind: 'membership', cursor: null, removeMember: true },
+      })
+      .where(eq(knowledgeConnectorMember.id, removedMember.id))
+    const onlyRemoved = row('only-the-removed-member')
+    const sharedWithOther = row('shared-with-other-member')
+    const neverObserved = row('never-observed')
+    await insertRows([onlyRemoved, sharedWithOther, neverObserved])
+    await observe([onlyRemoved.id, sharedWithOther.id])
+    await recordMemberObservations(db, otherMember.id, [sharedWithOther.id], members.runId)
+
+    const unobservedDocumentIds = new Set<string>()
+    expect(
+      await resumeMembershipRewrites({
+        connectorId: members.connectorId,
+        runId: members.runId,
+        deadlineAt: Date.now() + 60_000,
+        lease: createMemberSyncLease(members.connectorId, members.runId),
+        unobservedDocumentIds,
+      })
+    ).toBe(true)
+    const [completed] = await db
+      .select({ count: sql<number>`count(*)::int` })
+      .from(knowledgeConnectorMember)
+      .where(
+        and(
+          eq(knowledgeConnectorMember.connectorId, members.connectorId),
+          isNotNull(knowledgeConnectorMember.lastCompleteListingAt)
+        )
+      )
+    expect(completed.count).toBe(0)
+
+    expect(
+      await run({
+        allowRemoval: completed.count > 0,
+        unobservedDocumentIds: [...unobservedDocumentIds],
+      })
+    ).toEqual({ tombstoned: 1, resurrected: 0, purged: 0, finished: true })
+    const tombstoned = await tombstonedIds()
+    expect(tombstoned.has(onlyRemoved.id)).toBe(true)
+    expect(tombstoned.has(sharedWithOther.id)).toBe(false)
+    expect(tombstoned.has(neverObserved.id)).toBe(false)
+  })
+
+  it('finishes a pass within its page budget while listings re-stamp every observed document', async () => {
+    const pageBudget = MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN * 500
+    const total = pageBudget + 500
+    const runsPerPass = Math.ceil(total / pageBudget)
+    const firstInEveryOrder = {
+      ...row('walk-0000000'),
+      id: '00000000-0000-4000-8000-000000000000',
+    }
+    const rest = Array.from({ length: total - 1 }, (_, index) =>
+      row(`walk-${String(index + 1).padStart(7, '0')}`)
+    )
+    await insertRows([firstInEveryOrder, ...rest])
+    const all = [firstInEveryOrder, ...rest].map(({ id }) => id)
+    for (let offset = 0; offset < all.length; offset += 5000)
+      await observe(all.slice(offset, offset + 5000))
+    /** A listing stamps what it saw with its start; the document nobody observes keeps its old stamp. */
+    const relist = () =>
+      db
+        .update(document)
+        .set({ sourceSeenAt: new Date() })
+        .where(
+          and(
+            eq(document.connectorId, members.connectorId),
+            sql`EXISTS (SELECT 1 FROM knowledge_document_observation o WHERE o.document_id = ${document.id})`
+          )
+        )
+
+    expect(await run()).toMatchObject({ tombstoned: 0 })
+    expect(await savedCursor()).not.toBeNull()
+    await db
+      .delete(knowledgeDocumentObservation)
+      .where(eq(knowledgeDocumentObservation.documentId, firstInEveryOrder.id))
+    for (let pass = 1; pass < runsPerPass; pass++) {
+      await relist()
+      await run()
+    }
+    expect(await savedCursor()).toBeNull()
+    expect((await tombstonedIds()).has(firstInEveryOrder.id)).toBe(false)
+
+    for (let next = 0; next < runsPerPass; next++) {
+      await relist()
+      await run()
+    }
+    expect(await tombstonedIds()).toEqual(new Set([firstInEveryOrder.id]))
+  })
+
   it('continues past a full selected batch even if its observations change before UPDATE', async () => {
     const rows = Array.from({ length: 501 }, (_, index) => row(String(index)))
     await db.insert(document).values(rows)

apps/sim/lib/knowledge/connectors/member-observations.test.ts

@@ -261,7 +261,7 @@ describe('applyMemberDocumentLifecycle', () => {
     withLease: (fn) => fn(db),
     ...overrides,
   })
-  const pageRow = (id: string, live = true) => ({ id, seenAt: '2026-01-01 00:00:00', live })
+  const pageRow = (id: string) => ({ id, externalId: `ext-${id}` })
   /** The WHERE of every statement that targets documents by id, in call order. */
   const documentUpdateConditions = () =>
     dbChainMockFns.where.mock.calls
@@ -288,11 +288,7 @@ describe('applyMemberDocumentLifecycle', () => {
   })
 
   it('checks observations only for the live documents of one bounded page', async () => {
-    queueTableRows(schemaMock.document, [
-      pageRow('observed'),
-      pageRow('unobserved'),
-      pageRow('already-tombstoned', false),
-    ])
+    queueTableRows(schemaMock.document, [pageRow('observed'), pageRow('unobserved')])
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'unobserved' }])
 
     await expect(applyMemberDocumentLifecycle(lifecycleInput())).resolves.toEqual({
@@ -314,6 +310,17 @@ describe('applyMemberDocumentLifecycle', () => {
     expect(
       hasMockCondition(pageCondition, (node) => node.type === 'notExists' || node.type === 'exists')
     ).toBe(false)
+    expect(
+      hasMockCondition(
+        pageCondition,
+        (node) => node.type === 'isNull' && node.column === schemaMock.document.deletedAt
+      )
+    ).toBe(true)
+    /** An immutable key: `source_seen_at` moves on every listing, so a walk ordered by it never ends. */
+    expect(dbChainMockFns.orderBy).toHaveBeenCalledWith({
+      type: 'asc',
+      column: schemaMock.document.externalId,
+    })
     expect(dbChainMockFns.limit).toHaveBeenCalledWith(500)
     const [tombstone] = documentUpdateConditions()
     expect(updatedIds(tombstone)).toEqual(['observed', 'unobserved'])
@@ -322,9 +329,7 @@ describe('applyMemberDocumentLifecycle', () => {
   })
 
   it('stops after its page budget, saving where the next run resumes', async () => {
-    queueTableRows(schemaMock.knowledgeConnector, [
-      { cursor: { seenAt: '2025-12-31 00:00:00', id: 'previous-run' } },
-    ])
+    queueTableRows(schemaMock.knowledgeConnector, [{ cursor: { externalId: 'previous-run' } }])
     for (let page = 0; page < MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN; page++) {
       queueTableRows(
         schemaMock.document,
@@ -344,15 +349,18 @@ describe('applyMemberDocumentLifecycle', () => {
     const firstPage = dbChainMockFns.where.mock.calls
       .map(([condition]) => flattenMockConditions(condition))
       .find((nodes) => nodes.some((node) => node.left === schemaMock.document.connectorId))
-    expect(JSON.stringify(firstPage)).toContain('previous-run')
+    expect(firstPage).toContainEqual({
+      type: 'gt',
+      left: schemaMock.document.externalId,
+      right: 'previous-run',
+    })
     const cursors = dbChainMockFns.set.mock.calls
       .map(([value]) => value)
       .filter((value) => 'memberTombstoneCursor' in value)
     expect(cursors).toHaveLength(MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN)
     expect(cursors.at(-1)).toEqual({
       memberTombstoneCursor: {
-        seenAt: '2026-01-01 00:00:00',
-        id: `p${MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN - 1}-499`,
+        externalId: `ext-p${MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN - 1}-499`,
       },
     })
     expect(documentUpdateConditions()).toHaveLength(MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN)
@@ -380,12 +388,20 @@ describe('applyMemberDocumentLifecycle', () => {
     })
   })
 
-  it('removes nothing until a member has completed a listing', async () => {
+  it('before any completed listing, tombstones only what this run explicitly unobserved', async () => {
     queueTableRows(schemaMock.document, [])
-    await applyMemberDocumentLifecycle(
-      lifecycleInput({ allowRemoval: false, unobservedDocumentIds: ['d-1'] })
-    )
-    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'd-removed-member' }])
+    await expect(
+      applyMemberDocumentLifecycle(
+        lifecycleInput({ allowRemoval: false, unobservedDocumentIds: ['d-removed-member'] })
+      )
+    ).resolves.toEqual({ tombstoned: 1, resurrected: 0, purged: 0, finished: true })
+    const [targeted] = documentUpdateConditions()
+    expect(updatedIds(targeted)).toEqual(['d-removed-member'])
+    expect(hasMockCondition(targeted, (node) => node.type === 'notExists')).toBe(true)
+    /** Neither the absence reconcile nor the purge runs: absence alone still says nothing. */
+    expect(dbChainMockFns.update).not.toHaveBeenCalledWith(schemaMock.knowledgeConnector)
+    expect(documentUpdateConditions()).toHaveLength(1)
     expect(hardDeleteDocuments).not.toHaveBeenCalled()
   })
 

apps/sim/lib/knowledge/connectors/member-observations.ts

@@ -371,15 +371,17 @@ interface MemberDocumentLifecycleInput {
   withLease: <T>(fn: (tx: DbOrTx) => Promise<T>) => Promise<T>
   deadlineAt: number
   /**
-   * Whether absence of observers may hide or purge a document. False until at
-   * least one member has completed a listing: before that, nothing has been
-   * observed yet, so absence says nothing.
+   * Whether absence of observers may hide or purge a document the run has no
+   * explicit word on. False until at least one member has completed a listing:
+   * before that, nothing has been observed yet, so absence says nothing.
    */
   allowRemoval: boolean
   /**
    * Documents whose observations this run removed. Each is tombstoned if it has
-   * no observer left; absence arising any other way is found by the resumable
-   * reconcile.
+   * no observer left, whether or not `allowRemoval` holds: a removal is the
+   * source's or the directory's explicit word, not an absence, just as the
+   * stale-member sweep tombstones what it removes. Absence arising any other
+   * way is found by the resumable reconcile.
    */
   unobservedDocumentIds: Iterable<string>
 }
@@ -395,7 +397,7 @@ function unobservedLiveDocument(connectorId: string) {
   )
 }
 
-/** The order of `doc_connector_reconciliation_idx`, which every lifecycle page walks. */
+/** The order of `doc_connector_reconciliation_idx`, which the resurrection pages walk. */
 const seenOrder = sql`COALESCE(${document.sourceSeenAt}, '-infinity'::timestamp)`
 
 type TombstoneCursor = NonNullable<
@@ -432,13 +434,16 @@ async function tombstoneUnobserved(
  * The backstop for absence the run did not cause itself — a member row
  * deleted by an earlier run, a document restored from exclusion, a connector
  * whose first listing just completed, or a run that stopped between removing
- * observations and tombstoning. Walks the connector's documents in index
- * order, one page per statement, and checks observations only in the UPDATE
- * over that page's live ids: filtering the walk itself by observation lets
- * the LIMIT stop bounding it, and a select-list `EXISTS` can be planned as a
- * hash over every observation. Resumes from the cursor the previous run saved,
- * so each run's cost is bounded by the page budget rather than the
- * connector's size. Returns false when the deadline stopped it.
+ * observations and tombstoning. Walks the connector's live documents by
+ * external id through `doc_connector_external_id_idx`, one page per
+ * statement, and checks observations only in the UPDATE over that page's ids:
+ * filtering the walk itself by observation lets the LIMIT stop bounding it,
+ * and a select-list `EXISTS` can be planned as a hash over every observation.
+ * The key never changes for a document, unlike `source_seen_at`, which every
+ * listing rewrites: a walk ordered by it would chase the documents each run
+ * re-stamps and never reach the end of a pass. Resumes from the cursor the
+ * previous run saved, so each run's cost is bounded by the page budget rather
+ * than the connector's size. Returns false when the deadline stopped it.
  */
 async function reconcileUnobservedPages(
   input: MemberDocumentLifecycleInput,
@@ -454,29 +459,26 @@ async function reconcileUnobservedPages(
   for (let page = 0; page < MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN; page++) {
     if (Date.now() >= input.deadlineAt) return false
     await input.lease.beatIfDue()
+    /** Exclusion and archival are left to the UPDATE so every page is exactly one LIMIT of index entries. */
     const rows = await db
-      .select({
-        id: document.id,
-        seenAt: sql<string>`${seenOrder}::text`,
-        live: sql<boolean>`(${document.deletedAt} IS NULL)`,
-      })
+      .select({ id: document.id, externalId: document.externalId })
       .from(document)
       .where(
         and(
           eq(document.connectorId, connectorId),
-          eq(document.userExcluded, false),
-          isNull(document.archivedAt),
-          after
-            ? sql`(${seenOrder}, ${document.id}) > (${after.seenAt}::timestamp, ${after.id})`
-            : undefined
+          isNull(document.deletedAt),
+          isNotNull(document.externalId),
+          after ? gt(document.externalId, after.externalId) : undefined
         )
       )
-      .orderBy(seenOrder, asc(document.id))
+      .orderBy(asc(document.externalId))
       .limit(MATERIALIZE_BATCH_SIZE)
-    const candidates = rows.filter((row) => row.live).map((row) => row.id)
-    const last = rows.at(-1)
+    const candidates = rows.map((row) => row.id)
+    const lastExternalId = rows.at(-1)?.externalId
     const next: TombstoneCursor | null =
-      rows.length < MATERIALIZE_BATCH_SIZE || !last ? null : { seenAt: last.seenAt, id: last.id }
+      rows.length < MATERIALIZE_BATCH_SIZE || !lastExternalId
+        ? null
+        : { externalId: lastExternalId }
     if (Date.now() >= input.deadlineAt) return false
     const changed = await input.withLease(async (tx) => {
       const tombstoned =
@@ -508,7 +510,8 @@ async function reconcileUnobservedPages(
  * observation graph; visibility follows the active observers through the ACL.
  *
  * Tombstoning is driven by the documents whose observations this run removed,
- * then by a bounded slice of a resumable pass over the whole connector, so a
+ * then, once a member has completed a listing, by a bounded slice of a
+ * resumable pass over the whole connector, so a
  * run never evaluates every live document of a large connector in one
  * statement.
  *
@@ -528,11 +531,9 @@ export async function applyMemberDocumentLifecycle(
     purged: 0,
     finished: false,
   }
-  if (input.allowRemoval) {
-    const unobserved = [...new Set(input.unobservedDocumentIds)]
-    if (!(await tombstoneUnobserved(input, unobserved, now, result))) return result
-    if (!(await reconcileUnobservedPages(input, now, result))) return result
-  }
+  const unobserved = [...new Set(input.unobservedDocumentIds)]
+  if (!(await tombstoneUnobserved(input, unobserved, now, result))) return result
+  if (input.allowRemoval && !(await reconcileUnobservedPages(input, now, result))) return result
 
   let after: { id: string; seenAt: string } | undefined
   for (;;) {

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

@@ -2267,7 +2267,9 @@ export async function executeMemberSync(
          * Nobody has completed a listing yet — a connector that just entered
          * members mode, waiting for its first member to connect — so an
          * unobserved document says nothing about access and must not be
-         * tombstoned, let alone purged a week later.
+         * tombstoned, let alone purged a week later. What this run explicitly
+         * unobserved is tombstoned regardless, including after removing the
+         * last member that had completed a listing.
          */
         const [listed] = await db
           .select({ count: sql<number>`count(*)::int` })

packages/db/schema.ts

@@ -5914,14 +5914,11 @@ export const knowledgeConnector = pgTable(
      */
     accessRewritePending: boolean('access_rewrite_pending').notNull().default(false),
     /**
-     * Where the members-mode absence reconcile resumes: the last document it
-     * checked, in `doc_connector_reconciliation_idx` order. NULL starts a new
-     * pass from the beginning.
+     * Where the members-mode absence reconcile resumes: the external id of the
+     * last live document it checked, in `doc_connector_external_id_idx` order.
+     * NULL starts a new pass from the beginning.
      */
-    memberTombstoneCursor: jsonb('member_tombstone_cursor').$type<{
-      seenAt: string
-      id: string
-    }>(),
+    memberTombstoneCursor: jsonb('member_tombstone_cursor').$type<{ externalId: string }>(),
     /**
      * One of `active`, `pending`, `syncing`, `error`, `paused`, `disabled`.
      *