fix(knowledge): resume member scope renewal across runs

Accessible scopes are now listed page by page, and a renewal that does not
finish within its budget stores its channel-listing cursor and start time,
so the next run continues from there instead of re-reading the first pages
and never reaching channels past them. The watermark records when the whole
pass began, and an expired cursor restarts the pass once.

Author: waleedlatif1

apps/sim/connectors/slack/slack.test.ts

@@ -708,35 +708,34 @@ describe('Slack change detection and access scopes', () => {
   it('reports exactly the conversations each member can read, under the listing rules', async () => {
     const listScopes = slackConnector.listAccessibleScopes
     if (!listScopes) throw new Error('Slack must report access scopes')
+    const readAll = async (token: string, sourceConfig: Record<string, unknown>) => {
+      const prefixes: string[] = []
+      let cursor: string | undefined
+      do {
+        const page = await listScopes(token, sourceConfig, cursor)
+        prefixes.push(...page.prefixes)
+        cursor = page.nextCursor
+      } while (cursor)
+      return prefixes
+    }
     pageSize = 1
     expect(await listScopes('alice', {})).toEqual({
-      prefixes: [scope(GENERAL.id), scope(PRIVATE.id), scope(ARCHIVE.id)],
-      complete: true,
+      prefixes: [scope(GENERAL.id)],
+      nextCursor: '1',
     })
-    expect(await listScopes('bob', {})).toEqual({ prefixes: [scope(GENERAL.id)], complete: true })
+    expect(await readAll('alice', {})).toEqual([
+      scope(GENERAL.id),
+      scope(PRIVATE.id),
+      scope(ARCHIVE.id),
+    ])
+    expect(await readAll('bob', {})).toEqual([scope(GENERAL.id)])
     expect(
-      await listScopes('alice', { excludeChannels: '#general', includeArchived: 'false' })
-    ).toEqual({ prefixes: [scope(PRIVATE.id)], complete: true })
+      await readAll('alice', { excludeChannels: '#general', includeArchived: 'false' })
+    ).toEqual([scope(PRIVATE.id)])
     const listed = await listAll('alice', { maxMessages: 0 })
-    const { prefixes } = await listScopes('alice', {})
+    const prefixes = await readAll('alice', {})
     expect(
       listed.documents.every((doc) => prefixes.some((prefix) => doc.externalId.startsWith(prefix)))
     ).toBe(true)
   })
-
-  it('reports a conversation listing cut off at the page cap as incomplete', async () => {
-    const listScopes = slackConnector.listAccessibleScopes
-    if (!listScopes) throw new Error('Slack must report access scopes')
-    replacement = (call) =>
-      call.method === 'conversations.list'
-        ? {
-            ok: true,
-            channels: [GENERAL],
-            response_metadata: { next_cursor: String(Number(call.params.get('cursor') || 0) + 1) },
-          }
-        : undefined
-    const scopes = await listScopes('alice', {})
-    expect(scopes.complete).toBe(false)
-    expect(scopes.prefixes.length).toBeGreaterThan(0)
-  })
 })

apps/sim/connectors/slack/slack.ts

@@ -19,7 +19,7 @@ import {
   type ConnectorSourceFailureCategory,
 } from '@/connectors/source-error'
 import type {
-  AccessibleScopes,
+  AccessibleScopePage,
   ConnectorConfig,
   ExternalDocument,
   ExternalDocumentList,
@@ -43,8 +43,6 @@ const MAX_THREAD_PAGES = 200
 const MAX_USERNAME_CACHE_ENTRIES = 2000
 const MAX_CHANNEL_CACHE_ENTRIES = 2000
 const MAX_ROOT_VERSION_CACHE_ENTRIES = 50_000
-/** Conversation pages read when resolving a caller's reachable channels; each proves access on its own. */
-const MAX_SCOPE_CHANNEL_PAGES = 100
 /**
  * Threads with activity this recent are reread on every listing. A root reports new
  * replies and its own edits, but not an edit or deletion of an existing reply, which in
@@ -1008,26 +1006,19 @@ async function messageLink(
 }
 
 /**
- * The conversations the caller can read, under the same inclusion rules as a listing,
- * as the external-id prefix of their threads. Access is granted per conversation, so
- * a conversation listed here proves access to every thread in it. Stopping at the page
- * cap reports the listing incomplete, so the conversations past it stay due for renewal.
+ * One page of the conversations the caller can read, under the same inclusion rules as
+ * a listing, as the external-id prefix of their threads. Access is granted per
+ * conversation, so a conversation listed here proves access to every thread in it.
  */
 async function listAccessibleScopes(
   accessToken: string,
   sourceConfig: Record<string, unknown>,
+  cursor?: string,
   syncContext?: Record<string, unknown>
-): Promise<AccessibleScopes> {
+): Promise<AccessibleScopePage> {
   const { teamId } = await resolveWorkspace(accessToken, syncContext)
-  const prefixes: string[] = []
-  let cursor: string | undefined
-  for (let page = 0; page < MAX_SCOPE_CHANNEL_PAGES; page += 1) {
-    const { channels, nextCursor: next } = await listChannelPage(accessToken, sourceConfig, cursor)
-    for (const channel of channels) prefixes.push(channelScope(teamId, channel.id))
-    cursor = next
-    if (!cursor) break
-  }
-  return { prefixes, complete: !cursor }
+  const { channels, nextCursor: next } = await listChannelPage(accessToken, sourceConfig, cursor)
+  return { prefixes: channels.map((channel) => channelScope(teamId, channel.id)), nextCursor: next }
 }
 
 export const slackConnector: ConnectorConfig = {

apps/sim/connectors/types.ts

@@ -238,13 +238,13 @@ export type ExternalChange =
   | { kind: 'removed'; externalId: string }
 
 /**
- * The containers a caller can still read, as external-ID prefixes. Each listed
- * prefix proves access on its own; `complete` is false when the source stopped
- * before listing every container, so the rest are still due for renewal.
+ * One page of the containers a caller can still read, as external-ID prefixes.
+ * Each listed prefix proves access on its own; `nextCursor` continues the
+ * listing and is absent on its last page.
  */
-export interface AccessibleScopes {
+export interface AccessibleScopePage {
   prefixes: string[]
-  complete: boolean
+  nextCursor?: string
 }
 
 export interface ExternalChangeList {
@@ -582,14 +582,17 @@ export interface ConnectorConfig extends ConnectorMeta {
    * granted to whole containers (a Slack channel) rather than item by item. A
    * members-mode crawl renews the caller's observations under these prefixes
    * without relisting each item, so access stays fresh while a large listing is
-   * still in progress, and lapses for containers the caller has lost. Only
+   * still in progress, and lapses for containers the caller has lost. Paged, so
+   * a renewal can resume where an earlier run stopped; an expired cursor is
+   * recognised by {@link ConnectorConfig.isListingCursorInvalidError}. Only
    * meaningful alongside {@link ConnectorMeta.permissionScopedListing}.
    */
   listAccessibleScopes?: (
     accessToken: string,
     sourceConfig: Record<string, unknown>,
+    cursor?: string,
     syncContext?: Record<string, unknown>
-  ) => Promise<AccessibleScopes>
+  ) => Promise<AccessibleScopePage>
 
   /**
    * Opens the external directory whose groups this connector's mirrored ACLs

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

@@ -98,6 +98,8 @@ vi.mock('@/connectors/registry.server', () => ({
       listDocuments: mocks.list,
       getDocument: mocks.get,
       listAccessibleScopes: mocks.scopes,
+      isListingCursorInvalidError: (error: unknown) =>
+        error instanceof Error && error.message === 'cursor expired',
     },
     drive: {
       id: 'drive',
@@ -166,6 +168,7 @@ function arrange(
     organizationId?: string
     memberCheckpoint?: Record<string, unknown>
     scopeRenewedAt?: Date
+    scopeRenewal?: { cursor: string; startedAt: Date }
   } = {}
 ) {
   const connector = {
@@ -213,6 +216,8 @@ function arrange(
     : {
         ...member,
         scopeRenewedAt: options.scopeRenewedAt ?? null,
+        scopeRenewalCursor: options.scopeRenewal?.cursor ?? null,
+        scopeRenewalStartedAt: options.scopeRenewal?.startedAt ?? null,
         ...(options.memberCheckpoint ? { listingCheckpoint: options.memberCheckpoint } : {}),
       }
   const claims = options.members && !options.noDueMembers ? [[memberRow], []] : [[]]
@@ -358,6 +363,8 @@ describe('member engine with a dedicated content credential', () => {
         memberSyncedThrough: null,
         lastCompleteListingAt: null,
         scopeRenewedAt: null,
+        scopeRenewalCursor: null,
+        scopeRenewalStartedAt: null,
         listingCheckpoint: { kind: 'membership', cursor: null, removeMember: false },
         changeCursor: null,
         nextAttemptAt: expect.any(Date),
@@ -706,6 +713,8 @@ describe('member engine with a dedicated content credential', () => {
         memberSyncedThrough: null,
         lastCompleteListingAt: null,
         scopeRenewedAt: null,
+        scopeRenewalCursor: null,
+        scopeRenewalStartedAt: null,
         listingCheckpoint: { kind: 'membership', cursor: null, removeMember: false },
       })
     )
@@ -826,11 +835,13 @@ describe('member engine with a dedicated content credential', () => {
   })
 
   const scopeRenewal = () =>
-    dbChainMockFns.set.mock.calls.find(([value]) => 'scopeRenewedAt' in value)?.[0]
+    dbChainMockFns.set.mock.calls.find(
+      ([value]) => 'scopeRenewedAt' in value || 'scopeRenewalCursor' in value
+    )?.[0]
 
   it('renews member access by scope before listing and records when it finished', async () => {
     const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
-    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], complete: true })
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'] })
     mocks.renew.mockResolvedValue({ renewed: 3, finished: true })
     const result = await run()
     expect(result.error).toBeUndefined()
@@ -842,7 +853,26 @@ describe('member engine with a dedicated content credential', () => {
     expect(mocks.renew.mock.invocationCallOrder[0]).toBeLessThan(
       mocks.list.mock.invocationCallOrder[memberListing]
     )
-    expect(scopeRenewal()).toEqual({ scopeRenewedAt: expect.any(Date) })
+    expect(scopeRenewal()).toEqual({
+      scopeRenewedAt: expect.any(Date),
+      scopeRenewalCursor: null,
+      scopeRenewalStartedAt: null,
+    })
+  })
+
+  it('renews every page of scopes before recording the renewal', async () => {
+    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
+    mocks.scopes
+      .mockResolvedValueOnce({ prefixes: ['source:container-a:'], nextCursor: 'page-2' })
+      .mockResolvedValueOnce({ prefixes: ['source:container-b:'] })
+    mocks.renew.mockResolvedValue({ renewed: 2, finished: true })
+    expect((await run()).observationsRenewed).toBe(4)
+    expect(mocks.scopes.mock.calls.map((call) => call[2])).toEqual([undefined, 'page-2'])
+    expect(mocks.renew.mock.calls.map(([call]) => call.scopePrefixes)).toEqual([
+      ['source:container-a:'],
+      ['source:container-b:'],
+    ])
+    expect(scopeRenewal()).toMatchObject({ scopeRenewedAt: expect.any(Date) })
   })
 
   it('does not renew again while the last renewal is recent', async () => {
@@ -857,23 +887,56 @@ describe('member engine with a dedicated content credential', () => {
     expect(mocks.renew).not.toHaveBeenCalled()
   })
 
-  it('leaves an unfinished renewal due for the next run', async () => {
-    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
-    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], complete: true })
+  it('stores the cursor of an unfinished page so the next run reads it again', async () => {
+    const run = arrange({
+      connectorType: 'scoped_listing',
+      members: true,
+      contentFresh: true,
+      scopeRenewal: { cursor: 'page-7', startedAt: new Date('2026-09-01T00:00:00Z') },
+    })
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], nextCursor: 'page-8' })
     mocks.renew.mockResolvedValue({ renewed: 1000, finished: false })
     expect((await run()).observationsRenewed).toBe(1000)
-    expect(scopeRenewal()).toBeUndefined()
+    expect(mocks.scopes.mock.calls[0]?.[2]).toBe('page-7')
+    expect(scopeRenewal()).toEqual({
+      scopeRenewalCursor: 'page-7',
+      scopeRenewalStartedAt: new Date('2026-09-01T00:00:00Z'),
+    })
   })
 
-  it('leaves renewal due when the source stopped before listing every scope', async () => {
-    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
-    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], complete: false })
-    mocks.renew.mockResolvedValue({ renewed: 3, finished: true })
-    expect((await run()).observationsRenewed).toBe(3)
-    expect(mocks.renew).toHaveBeenCalledWith(
-      expect.objectContaining({ scopePrefixes: ['source:container-a:'] })
-    )
-    expect(scopeRenewal()).toBeUndefined()
+  it('resumes a stored pass and records its original start once every page is renewed', async () => {
+    const startedAt = new Date('2026-09-01T00:00:00Z')
+    const run = arrange({
+      connectorType: 'scoped_listing',
+      members: true,
+      contentFresh: true,
+      scopeRenewal: { cursor: 'page-7', startedAt },
+    })
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-z:'] })
+    mocks.renew.mockResolvedValue({ renewed: 1, finished: true })
+    await run()
+    expect(mocks.scopes.mock.calls.map((call) => call[2])).toEqual(['page-7'])
+    expect(scopeRenewal()).toEqual({
+      scopeRenewedAt: startedAt,
+      scopeRenewalCursor: null,
+      scopeRenewalStartedAt: null,
+    })
+  })
+
+  it('restarts a pass whose stored cursor expired', async () => {
+    const run = arrange({
+      connectorType: 'scoped_listing',
+      members: true,
+      contentFresh: true,
+      scopeRenewal: { cursor: 'page-7', startedAt: new Date('2026-09-01T00:00:00Z') },
+    })
+    mocks.scopes
+      .mockRejectedValueOnce(new Error('cursor expired'))
+      .mockResolvedValueOnce({ prefixes: ['source:container-a:'] })
+    mocks.renew.mockResolvedValue({ renewed: 1, finished: true })
+    await run()
+    expect(mocks.scopes.mock.calls.map((call) => call[2])).toEqual(['page-7', undefined])
+    expect(scopeRenewal()).toMatchObject({ scopeRenewalCursor: null })
   })
 
   it('still lists the member when their access scopes cannot be read', async () => {