fix(knowledge): prove the lease inside every ACL rewrite batch and harden source-modified-time parsing

Author: waleedlatif1

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

@@ -9,10 +9,12 @@ vi.mock('@/lib/knowledge/documents/service', () => ({
 }))
 
 import {
+  rewriteConnectorAcls,
   staleMemberWindowMs,
   sweepStaleMemberObservations,
 } from '@/lib/knowledge/connectors/member-observations'
 import { MEMBER_OBSERVATION_STALE_AFTER_HOURS } from '@/lib/knowledge/connectors/sync-limits'
+import { SyncLockLostException } from '@/lib/knowledge/connectors/sync-lock'
 
 const NOW = new Date('2026-09-01T12:00:00Z')
 const STALE_MEMBER = { id: 'm-1', connectorId: 'c-1', syncIntervalMinutes: 60 }
@@ -88,3 +90,33 @@ describe('sweepStaleMemberObservations', () => {
     expect(dbChainMockFns.delete).not.toHaveBeenCalled()
   })
 })
+
+describe('rewriteConnectorAcls', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('proves the lease inside each batch transaction before rewriting', async () => {
+    queueTableRows(schemaMock.knowledgeConnector, [{ id: 'c-1' }])
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'd-1' }])
+
+    await expect(
+      rewriteConnectorAcls('c-1', [], { lease: { stillHeld: () => 'held' as never } })
+    ).resolves.toBe(true)
+
+    expect(dbChainMockFns.transaction).toHaveBeenCalledOnce()
+    expect(dbChainMockFns.for).toHaveBeenCalledWith('share')
+    expect(dbChainMockFns.update).toHaveBeenCalledWith(schemaMock.document)
+  })
+
+  it('stops without writing once the lease is gone', async () => {
+    queueTableRows(schemaMock.knowledgeConnector, [])
+
+    await expect(
+      rewriteConnectorAcls('c-1', [], { lease: { stillHeld: () => 'lost' as never } })
+    ).rejects.toBeInstanceOf(SyncLockLostException)
+
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+  })
+})

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

@@ -27,9 +27,11 @@ import {
   MEMBER_TOMBSTONE_PURGE_DAYS,
 } from '@/lib/knowledge/connectors/sync-limits'
 import {
+  assertSyncLeaseHeldInTx,
   connectorIsLive,
   MEMBER_LOCKABLE_CONNECTOR_STATUSES,
   type SyncRunLease,
+  type SyncWriteLease,
 } from '@/lib/knowledge/connectors/sync-lock'
 import {
   type ConnectorSyncDeletionGuard,
@@ -184,28 +186,41 @@ const ACCESS_REWRITE_BATCH_SIZE = 1000
 export async function rewriteConnectorAcls(
   connectorId: string,
   target: readonly string[],
-  options: { deadlineAt?: number; beforeBatch?: () => Promise<void> } = {}
+  options: {
+    deadlineAt?: number
+    beforeBatch?: () => Promise<void>
+    /**
+     * The lease the caller holds on the connector, proved inside each batch's
+     * transaction: a heartbeat before the batch only says the lease was held
+     * then, and a run reclaimed mid-rewrite must not land an empty ACL over
+     * what its replacement has since materialised.
+     */
+    lease?: SyncWriteLease
+  } = {}
 ): Promise<boolean> {
   const mismatch =
     target.length === 0
       ? sql`cardinality(${document.acl}) > 0`
       : sql`${document.acl} <> ${textArrayLiteral(target)}`
   for (;;) {
     await options.beforeBatch?.()
-    const rewritten = await db
-      .update(document)
-      .set({ acl: [...target] })
-      .where(
-        eq(
-          document.id,
-          sql`ANY(ARRAY(
-            SELECT ${document.id} FROM ${document}
-            WHERE ${document.connectorId} = ${connectorId} AND ${mismatch}
-            LIMIT ${ACCESS_REWRITE_BATCH_SIZE}
-          ))`
+    const rewritten = await db.transaction(async (tx) => {
+      if (options.lease) await assertSyncLeaseHeldInTx(tx, connectorId, options.lease)
+      return tx
+        .update(document)
+        .set({ acl: [...target] })
+        .where(
+          eq(
+            document.id,
+            sql`ANY(ARRAY(
+              SELECT ${document.id} FROM ${document}
+              WHERE ${document.connectorId} = ${connectorId} AND ${mismatch}
+              LIMIT ${ACCESS_REWRITE_BATCH_SIZE}
+            ))`
+          )
         )
-      )
-      .returning({ id: document.id })
+        .returning({ id: document.id })
+    })
     if (rewritten.length < ACCESS_REWRITE_BATCH_SIZE) return true
     if (options.deadlineAt !== undefined && Date.now() >= options.deadlineAt) return false
   }

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

@@ -467,7 +467,10 @@ async function insertMemberSyncLog(runId: string, connectorId: string, startedAt
  * makes it visible again.
  */
 async function finishPendingAccessRewrite(run: MemberSyncRun): Promise<void> {
-  await rewriteConnectorAcls(run.connectorId, EMPTY_ACL, { beforeBatch: run.lease.beatIfDue })
+  await rewriteConnectorAcls(run.connectorId, EMPTY_ACL, {
+    beforeBatch: run.lease.beatIfDue,
+    lease: run.lease,
+  })
   await db
     .update(knowledgeConnector)
     .set({ accessRewritePending: false, updatedAt: new Date() })

apps/sim/lib/knowledge/connectors/source-modified-at.test.ts

@@ -46,8 +46,15 @@ describe('resolveSourceModifiedAt', () => {
     ).toBe('2026-08-01T00:00:00.000Z')
   })
 
-  it('rejects an invalid Date instance', () => {
+  it('rejects an invalid Date instance and a number outside the Date range', () => {
     expect(resolveSourceModifiedAt({ modifiedTime: new Date('not a date') })).toBeNull()
+    expect(resolveSourceModifiedAt({ modifiedTime: 1e20 })).toBeNull()
+  })
+
+  it('reads the last activity a chat space or channel reports', () => {
+    expect(resolveSourceModifiedAt({ lastActivity: '2026-08-30T10:00:00Z' })?.toISOString()).toBe(
+      '2026-08-30T10:00:00.000Z'
+    )
   })
 
   it('rejects placeholders and far-future values', () => {

apps/sim/lib/knowledge/connectors/source-modified-at.ts

@@ -16,23 +16,27 @@ const SOURCE_MODIFIED_AT_KEYS = [
   'updated',
   /** JSM requests: the time the request last changed status, the list endpoint's only change signal. */
   'statusDate',
+  /** Google Chat spaces and Teams channels: the latest message time, the listing's only change signal. */
+  'lastActivity',
 ] as const
 
 /** Earlier than any plausible document; guards against epoch-zero placeholders. */
 const EARLIEST_PLAUSIBLE_MS = Date.UTC(1990, 0, 1)
 /** A source clock a day ahead is skew; further ahead is a placeholder. */
 const FUTURE_TOLERANCE_MS = 24 * 60 * 60 * 1000
 
+/** A `Date` only when it holds a real instant; a finite number outside the Date range yields an invalid one. */
+function validDate(date: Date): Date | null {
+  return Number.isNaN(date.getTime()) ? null : date
+}
+
 function toDate(value: unknown): Date | null {
-  if (value instanceof Date) return Number.isNaN(value.getTime()) ? null : value
+  if (value instanceof Date) return validDate(value)
   if (typeof value === 'number' && Number.isFinite(value)) {
     /** Seconds-since-epoch values are far too small to be milliseconds after 1990. */
-    return new Date(value < 1e11 ? value * 1000 : value)
-  }
-  if (typeof value === 'string' && value.trim()) {
-    const parsed = new Date(value)
-    return Number.isNaN(parsed.getTime()) ? null : parsed
+    return validDate(new Date(value < 1e11 ? value * 1000 : value))
   }
+  if (typeof value === 'string' && value.trim()) return validDate(new Date(value))
   return null
 }
 

apps/sim/lib/knowledge/orchestration/connector-access.test.ts

@@ -20,6 +20,11 @@ const mocks = vi.hoisted(() => ({
   dispatchMemberSync: vi.fn(),
   memberAccessAvailable: vi.fn(),
   provision: vi.fn(),
+  rewriteAcls: vi.fn(),
+}))
+
+vi.mock('@/lib/knowledge/connectors/member-observations', () => ({
+  rewriteConnectorAcls: mocks.rewriteAcls,
 }))
 
 vi.mock('@sim/audit', () => ({
@@ -242,6 +247,7 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    mocks.rewriteAcls.mockResolvedValue(true)
     mocks.grant.mockResolvedValue(undefined)
     mocks.revoke.mockResolvedValue(undefined)
     mocks.dispatchSync.mockResolvedValue({ queued: true })
@@ -277,14 +283,21 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
     queueGroupRow('option-1')
     dbChainMockFns.returning
       .mockResolvedValueOnce([{ ...WORKSPACE_CONNECTOR, status: 'syncing', syncLockToken: 's-1' }])
-      /** The rewrite finds nothing left, the flip lands under the lease, then the release. */
-      .mockResolvedValueOnce([])
+      /** The flip lands under the lease, then the release. */
       .mockResolvedValueOnce([{ id: 'c-1' }])
       .mockResolvedValueOnce([{ ...MEMBERS_CONNECTOR, nextMemberSyncAt: new Date() }])
 
     const outcome = await switchTo({ accessMode: 'members', binding: BINDING })
 
     expect(outcome).toMatchObject({ success: true, changed: true })
+    /** The rewrite hides every document and proves the switch lease inside each batch. */
+    expect(mocks.rewriteAcls).toHaveBeenCalledWith(
+      'c-1',
+      [],
+      expect.objectContaining({
+        lease: expect.objectContaining({ stillHeld: expect.any(Function) }),
+      })
+    )
     expect(mocks.grant).toHaveBeenCalledWith(
       {
         workspaceId: 'ws-1',
@@ -294,7 +307,6 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
       },
       'admin-1'
     )
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(expect.objectContaining({ acl: [] }))
     /** The flip is written inside the group's row lock. */
     expect(dbChainMockFns.for).toHaveBeenCalledWith('update')
     expect(dbChainMockFns.transaction).toHaveBeenCalledOnce()
@@ -325,9 +337,9 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
   it('refuses the flip, and undoes the grant, when the option is gone by the time the group is locked', async () => {
     queueTableRows(schemaMock.knowledgeConnector, [WORKSPACE_CONNECTOR])
     queueGroupRow()
-    dbChainMockFns.returning
-      .mockResolvedValueOnce([{ ...WORKSPACE_CONNECTOR, status: 'syncing', syncLockToken: 's-1' }])
-      .mockResolvedValueOnce([])
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      { ...WORKSPACE_CONNECTOR, status: 'syncing', syncLockToken: 's-1' },
+    ])
 
     const outcome = await switchTo({ accessMode: 'members', binding: BINDING })
 
@@ -348,7 +360,6 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
     queueGroupRow('option-2')
     dbChainMockFns.returning
       .mockResolvedValueOnce([{ ...MEMBERS_CONNECTOR, status: 'syncing', syncLockToken: 's-1' }])
-      .mockResolvedValueOnce([])
       .mockResolvedValueOnce([{ id: 'c-1' }])
       .mockResolvedValueOnce([
         { ...MEMBERS_CONNECTOR, credentialGroupId: 'group-2', credentialGroupOptionId: 'option-2' },
@@ -384,16 +395,22 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
     queueTableRows(schemaMock.knowledgeConnector, [MEMBERS_CONNECTOR])
     dbChainMockFns.returning
       .mockResolvedValueOnce([{ ...MEMBERS_CONNECTOR, status: 'syncing', syncLockToken: 's-1' }])
-      /** The flip lands under the lease, then the rewrite finds nothing left, then the release. */
+      /** The flip lands under the lease, then the release. */
       .mockResolvedValueOnce([{ id: 'c-1' }])
-      .mockResolvedValueOnce([])
       .mockResolvedValueOnce([{ ...WORKSPACE_CONNECTOR, credentialId: 'cred-2' }])
 
     const outcome = await switchTo({ accessMode: 'workspace', credentialId: 'cred-2' })
 
     expect(outcome).toMatchObject({ success: true, changed: true })
     expect(dbChainMockFns.delete).toHaveBeenCalled()
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(expect.objectContaining({ acl: ['ws'] }))
+    /** Workspace access is restored under the switch lease, proved inside each batch. */
+    expect(mocks.rewriteAcls).toHaveBeenCalledWith(
+      'c-1',
+      ['ws'],
+      expect.objectContaining({
+        lease: expect.objectContaining({ stillHeld: expect.any(Function) }),
+      })
+    )
     expect(mocks.revoke).toHaveBeenCalledWith(
       { workspaceId: 'ws-1', credentialGroupId: 'group-1', connectorId: 'c-1' },
       'admin-1'
@@ -509,7 +526,6 @@ describe('performUpdateKnowledgeConnectorAccess', () => {
     queueGroupRow('option-1')
     dbChainMockFns.returning
       .mockResolvedValueOnce([{ ...WORKSPACE_CONNECTOR, status: 'paused' }])
-      .mockResolvedValueOnce([])
       .mockResolvedValueOnce([{ id: 'c-1' }])
       .mockResolvedValueOnce([{ ...MEMBERS_CONNECTOR, status: 'paused' }])
 

apps/sim/lib/knowledge/orchestration/connector-access.ts

@@ -335,6 +335,7 @@ export async function performUpdateKnowledgeConnectorAccess(
       try {
         const rewritten = await rewriteConnectorAcls(connectorId, EMPTY_ACL, {
           deadlineAt: deadlineAt,
+          lease: { stillHeld: () => switchLeaseHeld(connectorId, switchId) },
         })
         /**
          * The flip lands under the group's row lock, which the group's option
@@ -475,6 +476,7 @@ export async function performUpdateKnowledgeConnectorAccess(
     })
     const rewritten = await rewriteConnectorAcls(connectorId, WORKSPACE_ACL, {
       deadlineAt: deadlineAt,
+      lease: { stillHeld: () => switchLeaseHeld(connectorId, switchId) },
     })
     if (existing.credentialGroupId) {
       await revokeKnowledgeConnectorCredentialAccess(