fix(knowledge): keep scope renewal due when a channel listing is cut off

A Slack conversation listing that stops at its page cap now reports itself
incomplete, and the member's renewal watermark only advances once every
reachable channel has been listed and renewed.

Author: waleedlatif1

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

@@ -709,19 +709,34 @@ describe('Slack change detection and access scopes', () => {
     const listScopes = slackConnector.listAccessibleScopes
     if (!listScopes) throw new Error('Slack must report access scopes')
     pageSize = 1
-    expect(await listScopes('alice', {})).toEqual([
-      scope(GENERAL.id),
-      scope(PRIVATE.id),
-      scope(ARCHIVE.id),
-    ])
-    expect(await listScopes('bob', {})).toEqual([scope(GENERAL.id)])
+    expect(await listScopes('alice', {})).toEqual({
+      prefixes: [scope(GENERAL.id), scope(PRIVATE.id), scope(ARCHIVE.id)],
+      complete: true,
+    })
+    expect(await listScopes('bob', {})).toEqual({ prefixes: [scope(GENERAL.id)], complete: true })
     expect(
       await listScopes('alice', { excludeChannels: '#general', includeArchived: 'false' })
-    ).toEqual([scope(PRIVATE.id)])
+    ).toEqual({ prefixes: [scope(PRIVATE.id)], complete: true })
     const listed = await listAll('alice', { maxMessages: 0 })
-    const scopes = await listScopes('alice', {})
+    const { prefixes } = await listScopes('alice', {})
     expect(
-      listed.documents.every((doc) => scopes.some((prefix) => doc.externalId.startsWith(prefix)))
+      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

@@ -18,7 +18,12 @@ import {
   ConnectorSourceError,
   type ConnectorSourceFailureCategory,
 } from '@/connectors/source-error'
-import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types'
+import type {
+  AccessibleScopes,
+  ConnectorConfig,
+  ExternalDocument,
+  ExternalDocumentList,
+} from '@/connectors/types'
 import {
   BoundedLines,
   CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
@@ -1005,23 +1010,24 @@ 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.
+ * 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.
  */
 async function listAccessibleScopes(
   accessToken: string,
   sourceConfig: Record<string, unknown>,
   syncContext?: Record<string, unknown>
-): Promise<string[]> {
+): Promise<AccessibleScopes> {
   const { teamId } = await resolveWorkspace(accessToken, syncContext)
-  const scopes: string[] = []
+  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) scopes.push(channelScope(teamId, channel.id))
+    for (const channel of channels) prefixes.push(channelScope(teamId, channel.id))
     cursor = next
     if (!cursor) break
   }
-  return scopes
+  return { prefixes, complete: !cursor }
 }
 
 export const slackConnector: ConnectorConfig = {

apps/sim/connectors/types.ts

@@ -237,6 +237,16 @@ export type ExternalChange =
   | { kind: 'upsert'; externalId: string; document: ExternalDocument }
   | { 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.
+ */
+export interface AccessibleScopes {
+  prefixes: string[]
+  complete: boolean
+}
+
 export interface ExternalChangeList {
   changes: ExternalChange[]
   /**
@@ -579,7 +589,7 @@ export interface ConnectorConfig extends ConnectorMeta {
     accessToken: string,
     sourceConfig: Record<string, unknown>,
     syncContext?: Record<string, unknown>
-  ) => Promise<string[]>
+  ) => Promise<AccessibleScopes>
 
   /**
    * Opens the external directory whose groups this connector's mirrored ACLs

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

@@ -828,7 +828,7 @@ describe('member engine with a dedicated content credential', () => {
 
   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(['source:container-a:'])
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], complete: true })
     mocks.renew.mockResolvedValue({ renewed: 3, finished: true })
     const result = await run()
     expect(result.error).toBeUndefined()
@@ -857,12 +857,23 @@ describe('member engine with a dedicated content credential', () => {
 
   it('leaves an unfinished renewal due for the next run', async () => {
     const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
-    mocks.scopes.mockResolvedValue(['source:container-a:'])
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], complete: true })
     mocks.renew.mockResolvedValue({ renewed: 1000, finished: false })
     expect((await run()).observationsRenewed).toBe(1000)
     expect(scopeRenewal()).toBeUndefined()
   })
 
+  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('still lists the member when their access scopes cannot be read', async () => {
     const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
     mocks.scopes.mockRejectedValue(new Error('provider unavailable'))

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

@@ -1027,22 +1027,22 @@ async function renewMemberAccessScopes(input: {
   /** Observations under lost containers stay stale, so only the member's own watermark says renewal is due. */
   if (member.scopeRenewedAt && member.scopeRenewedAt > renewBefore) return
   try {
-    const scopePrefixes = await connectorConfig.listAccessibleScopes(
+    const scopes = await connectorConfig.listAccessibleScopes(
       await input.tokens.get(member.id),
       input.sourceConfig,
       input.syncContext
     )
     const renewal = await renewMemberObservationsInScopes({
       connectorId: run.connectorId,
       memberId: member.id,
-      scopePrefixes,
+      scopePrefixes: scopes.prefixes,
       renewBefore,
       deadlineAt: Math.min(run.deadlineAt, Date.now() + MEMBER_SCOPE_RENEWAL_BUDGET_MS),
       beforeBatch: run.lease.beatIfDue,
       withLease: (fn) => withMemberLease(run, fn),
     })
     run.result.observationsRenewed += renewal.renewed
-    if (renewal.finished) {
+    if (renewal.finished && scopes.complete) {
       await withMemberLease(run, (tx) =>
         tx
           .update(knowledgeConnectorMember)
@@ -1053,7 +1053,8 @@ async function renewMemberAccessScopes(input: {
     logger.info('Renewed member observations by access scope', {
       connectorId: run.connectorId,
       memberId: member.id,
-      scopes: scopePrefixes.length,
+      scopes: scopes.prefixes.length,
+      scopesComplete: scopes.complete,
       renewed: renewal.renewed,
       finished: renewal.finished,
     })