fix(knowledge): batch scope renewal and keep refreshed metadata

Scope renewal now gathers container pages into batches before scanning the
member's stale observations, so the scan runs once per batch instead of once
per source page, and an unfinished batch resumes from where it was read.
Content that hydrates unchanged under a new hash also refreshes its source
URL, modified time and tags, and the member sync log records how many
observations renewal kept fresh.

Author: waleedlatif1

apps/sim/background/knowledge-connector-member-sync.ts

@@ -68,6 +68,7 @@ export async function executeMemberSyncJob(payload: unknown) {
       unchanged: result.docsUnchanged,
       failed: result.docsFailed,
       observationsAdded: result.observationsAdded,
+      observationsRenewed: result.observationsRenewed,
       observationsRemoved: result.observationsRemoved,
       tombstoned: result.docsTombstoned,
       resurrected: result.docsResurrected,

apps/sim/lib/api/contracts/knowledge/connectors.ts

@@ -206,6 +206,8 @@ export const memberSyncLogDataSchema = z
     docsUnchanged: z.number(),
     docsHydratedOnce: z.number(),
     observationsAdded: z.number(),
+    /** Absent from responses served by older deployments. */
+    observationsRenewed: z.number().int().nonnegative().optional(),
     observationsRemoved: z.number(),
     docsTombstoned: z.number(),
     docsResurrected: z.number(),

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

@@ -71,6 +71,7 @@ vi.mock('@/lib/knowledge/connectors/sync-persistence', () => ({
   persistSkippedDocuments: vi.fn(async () => []),
   persistSourceDocumentFailures: mocks.persistFailures,
   persistHashOnlyUpdates: vi.fn(async () => []),
+  resolveSourceMetadataFields: vi.fn(() => ({ sourceUrl: null, sourceModifiedAt: null })),
 }))
 vi.mock('@/lib/knowledge/documents/service', () => ({
   isTriggerAvailable: () => true,
@@ -117,6 +118,7 @@ vi.mock('@/connectors/registry.server', () => ({
   },
 }))
 
+import { ProviderCapacityDeferredError } from '@/lib/core/rate-limiter/provider-capacity-error'
 import {
   CredentialGroupCredentialCursorNotFoundError,
   loadScopedAccountsCredentialListContext,
@@ -126,7 +128,10 @@ import {
   listingFingerprint,
 } from '@/lib/knowledge/connectors/listing-checkpoint'
 import { executeMemberSync } from '@/lib/knowledge/connectors/member-sync-engine'
-import { SOURCE_CONTENT_ERROR } from '@/lib/knowledge/connectors/sync-limits'
+import {
+  MEMBER_SCOPE_RENEWAL_PREFIX_BATCH,
+  SOURCE_CONTENT_ERROR,
+} from '@/lib/knowledge/connectors/sync-limits'
 
 const serviceDocument: ExternalDocument = {
   externalId: 'file-shared',
@@ -858,23 +863,59 @@ describe('member engine with a dedicated content credential', () => {
       scopeRenewalCursor: null,
       scopeRenewalStartedAt: null,
     })
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({ completedAt: expect.any(Date), observationsRenewed: 3 })
+    )
   })
 
-  it('renews every page of scopes before recording the renewal', async () => {
+  it('records renewed observations on the log of a deferred run', async () => {
+    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'] })
+    mocks.renew.mockResolvedValue({ renewed: 2, finished: true })
+    mocks.list.mockRejectedValue(new ProviderCapacityDeferredError('admission_unavailable'))
+    expect((await run()).deferred).toBeDefined()
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({ completedAt: expect.any(Date), observationsRenewed: 2 })
+    )
+  })
+
+  it('gathers scope pages into one pass over the stale observations', 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((await run()).observationsRenewed).toBe(2)
     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:'],
+      ['source:container-a:', 'source:container-b:'],
     ])
     expect(scopeRenewal()).toMatchObject({ scopeRenewedAt: expect.any(Date) })
   })
 
+  it('renews a full batch of scopes before reading more, and resumes an unfinished batch', async () => {
+    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
+    const fullBatch = Array.from(
+      { length: MEMBER_SCOPE_RENEWAL_PREFIX_BATCH },
+      (_, index) => `source:container-${index}:`
+    )
+    mocks.scopes
+      .mockResolvedValueOnce({ prefixes: fullBatch, nextCursor: 'page-2' })
+      .mockResolvedValueOnce({ prefixes: ['source:container-last:'] })
+    mocks.renew
+      .mockResolvedValueOnce({ renewed: 5, finished: true })
+      .mockResolvedValueOnce({ renewed: 1, finished: false })
+    expect((await run()).observationsRenewed).toBe(6)
+    expect(mocks.renew.mock.calls.map(([call]) => call.scopePrefixes.length)).toEqual([
+      MEMBER_SCOPE_RENEWAL_PREFIX_BATCH,
+      1,
+    ])
+    expect(scopeRenewal()).toEqual({
+      scopeRenewalCursor: 'page-2',
+      scopeRenewalStartedAt: expect.any(Date),
+    })
+  })
+
   it('does not renew again while the last renewal is recent', async () => {
     const run = arrange({
       connectorType: 'scoped_listing',
@@ -894,7 +935,7 @@ describe('member engine with a dedicated content credential', () => {
       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.scopes.mockResolvedValue({ prefixes: ['source:container-a:'] })
     mocks.renew.mockResolvedValue({ renewed: 1000, finished: false })
     expect((await run()).observationsRenewed).toBe(1000)
     expect(mocks.scopes.mock.calls[0]?.[2]).toBe('page-7')
@@ -923,6 +964,16 @@ describe('member engine with a dedicated content credential', () => {
     })
   })
 
+  it('stops a scope listing whose cursor does not advance', async () => {
+    const run = arrange({ connectorType: 'scoped_listing', members: true, contentFresh: true })
+    mocks.scopes.mockResolvedValue({ prefixes: ['source:container-a:'], nextCursor: 'page-2' })
+    mocks.renew.mockResolvedValue({ renewed: 1, finished: true })
+    expect((await run()).error).toBeUndefined()
+    expect(mocks.scopes).toHaveBeenCalledTimes(2)
+    expect(mocks.renew).not.toHaveBeenCalled()
+    expect(mocks.observe).toHaveBeenCalled()
+  })
+
   it('restarts a pass whose stored cursor expired', async () => {
     const run = arrange({
       connectorType: 'scoped_listing',

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

@@ -76,6 +76,7 @@ import {
   MEMBER_FULL_RECRAWL_MINUTES,
   MEMBER_SCOPE_RENEW_AFTER_MS,
   MEMBER_SCOPE_RENEWAL_BUDGET_MS,
+  MEMBER_SCOPE_RENEWAL_PREFIX_BATCH,
   MEMBER_SUSPENDED_PURGE_DAYS,
   MEMBER_SYNC_MAX_PAGES_PER_MEMBER,
   MEMBER_SYNC_SOFT_BUDGET_SECONDS,
@@ -1039,6 +1040,9 @@ async function renewMemberAccessScopes(input: {
   /** A resumed pass keeps its start, so the watermark never claims more than the whole pass renewed. */
   const passStartedAt = member.scopeRenewalStartedAt ?? now
   let cursor = member.scopeRenewalCursor ?? undefined
+  /** Where the prefixes not yet renewed were read from; an unfinished pass resumes there. */
+  let batchCursor = cursor
+  let pending: string[] = []
   let restartedExpiredCursor = false
   let scopes = 0
   let renewed = 0
@@ -1068,23 +1072,30 @@ async function renewMemberAccessScopes(input: {
         )
           throw error
         cursor = undefined
+        batchCursor = undefined
+        pending = []
         restartedExpiredCursor = true
         continue
       }
+      if (page.nextCursor && page.nextCursor === cursor)
+        throw new Error('Access scope pagination did not advance')
       scopes += page.prefixes.length
+      pending.push(...page.prefixes)
+      cursor = page.nextCursor
+      if (cursor && pending.length < MEMBER_SCOPE_RENEWAL_PREFIX_BATCH) continue
       const renewal = await renewMemberObservationsInScopes({
         connectorId: run.connectorId,
         memberId: member.id,
-        scopePrefixes: page.prefixes,
+        scopePrefixes: pending,
         renewBefore,
         deadlineAt,
         beforeBatch: run.lease.beatIfDue,
         withLease: (fn) => withMemberLease(run, fn),
       })
       renewed += renewal.renewed
-      /** An unfinished page is read again next time, from the cursor that produced it. */
       if (!renewal.finished) break
-      cursor = page.nextCursor
+      pending = []
+      batchCursor = cursor
       if (!cursor) {
         finished = true
         await saveProgress({
@@ -1097,7 +1108,7 @@ async function renewMemberAccessScopes(input: {
     }
     if (!finished)
       await saveProgress({
-        scopeRenewalCursor: cursor ?? null,
+        scopeRenewalCursor: batchCursor ?? null,
         scopeRenewalStartedAt: passStartedAt,
       })
   } catch (error) {
@@ -1626,6 +1637,7 @@ async function completeMemberSync(
         docsUnchanged: result.docsUnchanged,
         docsHydratedOnce: result.docsHydratedOnce,
         observationsAdded: result.observationsAdded,
+        observationsRenewed: result.observationsRenewed,
         observationsRemoved: result.observationsRemoved,
         docsTombstoned: result.docsTombstoned,
         docsResurrected: result.docsResurrected,
@@ -1679,6 +1691,7 @@ async function failMemberSyncLog(runId: string, result: MemberSyncResult, errorM
       docsUnchanged: result.docsUnchanged,
       docsHydratedOnce: result.docsHydratedOnce,
       observationsAdded: result.observationsAdded,
+      observationsRenewed: result.observationsRenewed,
       observationsRemoved: result.observationsRemoved,
       docsTombstoned: result.docsTombstoned,
       docsResurrected: result.docsResurrected,

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

@@ -731,6 +731,35 @@ describe('persistHashOnlyUpdates', () => {
     ).toEqual({ type: 'update', existingId: 'doc-1' })
   })
 
+  it('writes refreshed source metadata alongside an equivalent content hash', async () => {
+    const { persistHashOnlyUpdates } = await import('@/lib/knowledge/connectors/sync-persistence')
+    queueTableRows(schemaMock.knowledgeBase, [{ id: 'kb-1' }])
+    queueTableRows(schemaMock.knowledgeConnector, [{ id: 'connector-1' }])
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'doc-1' }])
+    const sourceModifiedAt = new Date('2026-09-08T12:00:00Z')
+
+    await persistHashOnlyUpdates(
+      'kb-1',
+      'connector-1',
+      [
+        {
+          existingId: 'doc-1',
+          externalId: 'thread-1',
+          contentHash: 'slack-thread:v5:version:text',
+          sourceMetadata: { sourceUrl: null, sourceModifiedAt, date1: sourceModifiedAt },
+        },
+      ],
+      lease
+    )
+
+    expect(dbChainMockFns.set).toHaveBeenCalledWith({
+      contentHash: 'slack-thread:v5:version:text',
+      sourceUrl: null,
+      sourceModifiedAt,
+      date1: sourceModifiedAt,
+    })
+  })
+
   it('commits live retry hashes when another document is no longer a connector target', async () => {
     const { persistHashOnlyUpdates } = await import('@/lib/knowledge/connectors/sync-persistence')
     queueTableRows(schemaMock.knowledgeBase, [{ id: 'kb-1' }])

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

@@ -85,6 +85,13 @@ export const MEMBER_SCOPE_RENEW_AFTER_MS = SOURCE_ACL_MAX_AGE_MS / 2
 /** How much of a run one member's scope renewal may use before its listing starts. */
 export const MEMBER_SCOPE_RENEWAL_BUDGET_MS = 10 * 60 * 1000
 
+/**
+ * Container prefixes gathered from the source before one pass over the member's stale
+ * observations renews them, so that pass runs once per this many containers rather than
+ * once per source page.
+ */
+export const MEMBER_SCOPE_RENEWAL_PREFIX_BATCH = 5000
+
 /** Pages applied per member before its durable feed cursor is saved for continuation. */
 export const MEMBER_SYNC_MAX_PAGES_PER_MEMBER = 25
 

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

@@ -435,16 +435,41 @@ export async function persistSkippedDocuments(
   return persisted
 }
 
+/** Source-derived fields that can change without the indexed text changing. */
+export type SourceMetadataFields = Partial<DocumentTags> & {
+  sourceUrl: string | null
+  sourceModifiedAt: Date | null
+}
+
+/** The source-derived fields a refreshed document carries, as a full update writes them. */
+export function resolveSourceMetadataFields(
+  connectorType: string,
+  extDoc: Pick<ExternalDocument, 'sourceUrl' | 'metadata'>,
+  sourceConfig: Record<string, unknown>
+): SourceMetadataFields {
+  return {
+    sourceUrl: extDoc.sourceUrl ?? null,
+    sourceModifiedAt: resolveSourceModifiedAt(extDoc.metadata),
+    ...(extDoc.metadata ? resolveTagMapping(connectorType, extDoc.metadata, sourceConfig) : {}),
+  }
+}
+
 /**
  * Persists only a new hash for existing documents, leaving indexed content and
  * processing state as they are: a connector-owned retry hash for a skipped
  * refresh, so unchanged listing metadata still re-enters hydration, or the
- * current hash of content that hydration found unchanged under an older one.
+ * current hash of content that hydration found unchanged under an older one,
+ * together with that hydration's source metadata, which can move on its own.
  */
 export async function persistHashOnlyUpdates(
   knowledgeBaseId: string,
   connectorId: string,
-  updates: Array<{ existingId: string; externalId: string; contentHash: string }>,
+  updates: Array<{
+    existingId: string
+    externalId: string
+    contentHash: string
+    sourceMetadata?: SourceMetadataFields
+  }>,
   lease: SyncWriteLease
 ): Promise<string[]> {
   if (updates.length === 0) return []
@@ -461,7 +486,7 @@ export async function persistHashOnlyUpdates(
     for (const update of updates) {
       const persisted = await tx
         .update(document)
-        .set({ contentHash: update.contentHash })
+        .set({ contentHash: update.contentHash, ...update.sourceMetadata })
         .where(connectorDocumentSyncTarget(update.existingId, knowledgeBaseId, connectorId))
         .returning({ id: document.id })
       if (persisted.length === 0) {

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

@@ -9,6 +9,13 @@ const mocks = vi.hoisted(() => ({
   update: vi.fn(),
   triggerAvailable: vi.fn(),
   persistHashes: vi.fn(),
+  sourceMetadata: vi.fn(
+    (_connectorType: string, doc: Pick<ExternalDocument, 'sourceUrl' | 'metadata'>) => ({
+      sourceUrl: doc.sourceUrl ?? null,
+      sourceModifiedAt: null,
+      date1: doc.metadata?.lastActivity,
+    })
+  ),
   dispatch: vi.fn<(documents: DocumentData[]) => Promise<{ accepted: number; failed: number }>>(),
 }))
 
@@ -17,6 +24,7 @@ vi.mock('@/lib/knowledge/connectors/sync-persistence', () => ({
   updateDocument: mocks.update,
   persistSkippedDocuments: vi.fn(),
   persistHashOnlyUpdates: mocks.persistHashes,
+  resolveSourceMetadataFields: mocks.sourceMetadata,
 }))
 vi.mock('@/lib/knowledge/documents/service', () => ({
   isTriggerAvailable: mocks.triggerAvailable,
@@ -250,13 +258,15 @@ describe('processDocOps unchanged content under a new hash', () => {
     input.hydration.getDocument = vi.fn(async () => ({
       ...sourceDocument('thread'),
       contentHash: hydratedHash,
+      sourceUrl: 'https://source.fixture.test/thread',
+      metadata: { lastActivity: '2026-09-08T12:00:00.000Z' },
     }))
     input.matchContentHash = (candidate, stored) =>
       candidate.split(':').at(-1) === stored.split(':').at(-1) ? 'equivalent' : 'stale'
     return input
   }
 
-  it('advances only the stored hash when the connector finds the same text', async () => {
+  it('advances the stored hash and source metadata when the connector finds the same text', async () => {
     const input = refreshOf('legacy:text-a', 'version:2:text-a')
     await expect(processDocOps(input)).resolves.toBe(true)
     expect(mocks.update).not.toHaveBeenCalled()
@@ -265,7 +275,18 @@ describe('processDocOps unchanged content under a new hash', () => {
     expect(mocks.persistHashes).toHaveBeenCalledWith(
       'knowledge-base',
       'connector',
-      [{ existingId: 'document', externalId: 'thread', contentHash: 'version:2:text-a' }],
+      [
+        {
+          existingId: 'document',
+          externalId: 'thread',
+          contentHash: 'version:2:text-a',
+          sourceMetadata: {
+            sourceUrl: 'https://source.fixture.test/thread',
+            sourceModifiedAt: null,
+            date1: '2026-09-08T12:00:00.000Z',
+          },
+        },
+      ],
       input.lease
     )
   })