fix(knowledge): grant member access as each batch is indexed instead of at the end of the run (#7418)

* fix(knowledge): grant member access as each batch is indexed instead of at the end of the run

* perf(connectors): hydrate deferred mail and chat documents in batches instead of one at a time

* fix(knowledge): grant access after dispatch, include skipped rows, and keep the grant best-effort

* fix(knowledge): cast the stale-sweep cutoff so the member-sync scheduler stops failing

* fix(connectors): cap assembled mail and chat text at one shared bound and advertise it as the hydration estimate

* fix(google-chat): report the number of messages the capped transcript actually indexed

Author: waleedlatif1

apps/sim/connectors/gmail/gmail.ts

@@ -4,6 +4,8 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document
 import { DEFAULT_MAX_THREADS, gmailConnectorMeta } from '@/connectors/gmail/meta'
 import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types'
 import {
+  BoundedLines,
+  CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
   htmlToPlainText,
   joinTagArray,
   parseDefaultedUnlimitedSafeInteger,
@@ -323,21 +325,17 @@ function formatThread(thread: GmailThread): {
   }
   const labelIds = [...labelIdSet]
 
-  const lines: string[] = []
-  lines.push(`Subject: ${subject}`)
-  lines.push(`From: ${from}`)
+  const lines = new BoundedLines()
+  lines.push(`Subject: ${subject}`, `From: ${from}`)
   if (to) lines.push(`To: ${to}`)
-  lines.push(`Messages: ${messages.length}`)
-  lines.push('')
+  lines.push(`Messages: ${messages.length}`, '')
 
   for (const msg of messages) {
     const msgFrom = getHeader(msg.payload, 'From') || 'Unknown'
     const msgDate = getHeader(msg.payload, 'Date') || ''
     const body = msg.payload ? extractBody(msg.payload) : ''
 
-    lines.push(`--- ${msgFrom} (${msgDate}) ---`)
-    lines.push(body.trim())
-    lines.push('')
+    if (!lines.push(`--- ${msgFrom} (${msgDate}) ---`, body.trim(), '')) break
   }
 
   const firstDate = firstMessage.internalDate
@@ -348,7 +346,7 @@ function formatThread(thread: GmailThread): {
     : undefined
 
   return {
-    content: lines.join('\n').trim(),
+    content: lines.join().trim(),
     subject,
     metadata: {
       from,
@@ -412,6 +410,7 @@ function threadToStub(thread: {
     title: thread.snippet || 'Untitled Thread',
     content: '',
     contentDeferred: true,
+    estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
     mimeType: 'text/plain',
     sourceUrl: threadUrl(thread.id),
     contentHash: `gmail:${thread.id}:${thread.historyId ?? ''}`,

apps/sim/connectors/google-chat/google-chat.ts

@@ -9,7 +9,7 @@ import {
   SPACES_PAGE_SIZE,
 } from '@/connectors/google-chat/meta'
 import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types'
-import { parseTagDate } from '@/connectors/utils'
+import { BoundedLines, CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, parseTagDate } from '@/connectors/utils'
 
 const logger = createLogger('GoogleChatConnector')
 
@@ -218,6 +218,7 @@ function spaceToStub(
     title: spaceTitle(space),
     content: '',
     contentDeferred: true,
+    estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
     mimeType: 'text/plain',
     sourceUrl: space.spaceUri,
     contentHash: buildContentHash(space, maxMessages, lookbackDays, syncContext),
@@ -323,37 +324,28 @@ function senderLabel(sender: ChatUser | undefined): string {
  * that belongs in a knowledge base is synced through the Google Drive connector
  * instead, which already handles size caps, OCR, and format parsing.
  */
-function formatSpaceContent(space: Space, messages: ChatMessage[]): string {
-  const parts: string[] = [`Space: ${spaceTitle(space)}`]
+function formatSpaceContent(
+  space: Space,
+  messages: ChatMessage[]
+): { content: string; messageCount: number } {
+  const parts = new BoundedLines()
+  parts.push(`Space: ${spaceTitle(space)}`)
   const description = space.spaceDetails?.description?.trim()
   if (description) parts.push(`Description: ${description}`)
   const guidelines = space.spaceDetails?.guidelines?.trim()
   if (guidelines) parts.push(`Guidelines: ${guidelines}`)
 
-  const lines: string[] = []
+  let messageCount = 0
   for (const message of messages) {
     const text = message.text?.trim() || message.fallbackText?.trim()
     if (!text) continue
+    if (messageCount === 0) parts.push('', '--- Messages ---')
     const timestamp = message.createTime ?? ''
-    lines.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`)
-  }
-
-  if (lines.length > 0) {
-    parts.push('')
-    parts.push('--- Messages ---')
-    parts.push(...lines)
+    if (!parts.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`)) break
+    messageCount += 1
   }
 
-  return parts.join('\n')
-}
-
-/** Number of messages that actually contributed text to the transcript. */
-function countIndexedMessages(messages: ChatMessage[]): number {
-  let count = 0
-  for (const message of messages) {
-    if (message.text?.trim() || message.fallbackText?.trim()) count++
-  }
-  return count
+  return { content: parts.join(), messageCount }
 }
 
 export const googleChatConnector: ConnectorConfig = {
@@ -476,12 +468,12 @@ export const googleChatConnector: ConnectorConfig = {
      * leave a previously indexed transcript in place after the space was cleared
      * or `lookbackDays` was tightened past every message.
      */
-    const messageCount = countIndexedMessages(messages)
+    const { content, messageCount } = formatSpaceContent(space, messages)
     const stub = spaceToStub(space, maxMessages, lookbackDays, syncContext)
 
     return {
       ...stub,
-      content: formatSpaceContent(space, messages),
+      content,
       contentDeferred: false,
       metadata: { ...stub.metadata, messageCount },
     }

apps/sim/connectors/outlook/outlook.ts

@@ -4,6 +4,8 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document
 import { DEFAULT_MAX_CONVERSATIONS, outlookConnectorMeta } from '@/connectors/outlook/meta'
 import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types'
 import {
+  BoundedLines,
+  CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
   htmlToPlainText,
   isListingScopeUnavailableError,
   listingRequestError,
@@ -563,24 +565,20 @@ function formatConversation(
   const from = formatRecipient(first.from)
   const to = first.toRecipients?.map(formatRecipient).join(', ') || ''
 
-  const lines: string[] = []
-  lines.push(`Subject: ${subject}`)
-  lines.push(`From: ${from}`)
+  const lines = new BoundedLines()
+  lines.push(`Subject: ${subject}`, `From: ${from}`)
   if (to) lines.push(`To: ${to}`)
-  lines.push(`Messages: ${sorted.length}`)
-  lines.push('')
+  lines.push(`Messages: ${sorted.length}`, '')
 
   for (const msg of sorted) {
     const msgFrom = formatRecipient(msg.from)
     const msgDate = msg.receivedDateTime || ''
     const body = extractBodyText(msg.body)
 
-    lines.push(`--- ${msgFrom} (${msgDate}) ---`)
-    lines.push(body.trim())
-    lines.push('')
+    if (!lines.push(`--- ${msgFrom} (${msgDate}) ---`, body.trim(), '')) break
   }
 
-  const content = lines.join('\n').trim()
+  const content = lines.join().trim()
   if (!content) return null
 
   const categories = new Set<string>()
@@ -805,6 +803,7 @@ export const outlookConnector: ConnectorConfig = {
         title: subject,
         content: '',
         contentDeferred: true,
+        estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
         mimeType: 'text/plain',
         sourceUrl,
         contentHash: `outlook:${convId}:${lastDate}`,

apps/sim/connectors/types.ts

@@ -63,6 +63,14 @@ export interface ExternalDocument {
   skippedRetryContentHash?: string
   /** When true, content is empty and will be fetched via getDocument for new/changed docs only */
   contentDeferred?: boolean
+  /**
+   * How large the deferred content is expected to be, in bytes, when the
+   * listing cannot know exactly. Bounds how many deferred documents hydrate
+   * at once: without it a deferred document is assumed to be as large as the
+   * whole in-flight budget and hydrates alone, which turns a mailbox crawl
+   * into one thread at a time.
+   */
+  estimatedBytes?: number
   /**
    * When set, the document was intentionally not indexed (e.g. it exceeds the
    * connector's size limit). The sync engine records it as a `failed` document

apps/sim/connectors/utils.test.ts

@@ -63,6 +63,7 @@ import { typeformConnector } from '@/connectors/typeform/typeform'
 import {
   appendPendingMicrosoftGraphFolders,
   assertMicrosoftGraphNextLink,
+  BoundedLines,
   ConnectorFileTooLargeError,
   ConnectorListingScopeUnavailableError,
   decodeMicrosoftGraphTraversalCursor,
@@ -1639,3 +1640,26 @@ describe('isSkippableMicrosoftGraphFolderError', () => {
     expect(isSkippableMicrosoftGraphFolderError(new Error('500'), perMember, false)).toBe(false)
   })
 })
+
+describe('BoundedLines', () => {
+  it('joins everything when the text fits', () => {
+    const lines = new BoundedLines(64)
+    expect(lines.push('Subject: hi', '')).toBe(true)
+    expect(lines.push('--- a ---', 'body')).toBe(true)
+    expect(lines.join()).toBe('Subject: hi\n\n--- a ---\nbody')
+  })
+
+  it('refuses a record that would cross the ceiling, whole, and says so in the output', () => {
+    const lines = new BoundedLines(20)
+    expect(lines.push('first')).toBe(true)
+    expect(lines.push('--- header ---', 'a long body')).toBe(false)
+    expect(lines.push('x')).toBe(false)
+    expect(lines.join()).toBe('first\n[Truncated: the indexed text reached the size limit]')
+  })
+
+  it('counts encoded bytes, not characters', () => {
+    const lines = new BoundedLines(6)
+    expect(lines.push('éé')).toBe(true)
+    expect(lines.push('é')).toBe(false)
+  })
+})

apps/sim/connectors/utils.ts

@@ -795,3 +795,50 @@ export function isSkippableMicrosoftGraphFolderError(
 ): boolean {
   return !isRootFolder && isListingScopeUnavailableError(error) && isPerMemberListing(syncContext)
 }
+
+/**
+ * Ceiling for a document a connector assembles from many source records (a
+ * mail thread, a chat transcript) rather than downloads as one file. Formatters
+ * enforce it through `BoundedLines`, and listings advertise it through
+ * `ExternalDocument.estimatedBytes`, so the sync engine plans hydration around
+ * a bound it can rely on: five such documents fit its 64 MiB in-flight budget
+ * and hydrate together instead of one at a time.
+ */
+export const CONNECTOR_TEXT_DOCUMENT_MAX_BYTES = 12 * 1024 * 1024
+
+const TRUNCATION_NOTICE = '[Truncated: the indexed text reached the size limit]'
+
+/**
+ * Accumulates newline-joined text under a byte ceiling. A record is appended
+ * whole or not at all, so a truncated document never ends mid-message, and the
+ * output carries a notice when something was left out.
+ */
+export class BoundedLines {
+  private readonly lines: string[] = []
+  private bytes = 0
+  private truncated = false
+
+  constructor(private readonly maxBytes = CONNECTOR_TEXT_DOCUMENT_MAX_BYTES) {}
+
+  /**
+   * Appends the lines together when they fit; otherwise marks the document
+   * truncated, appends nothing, and returns false so the caller stops.
+   */
+  push(...lines: string[]): boolean {
+    if (this.truncated) return false
+    let size = 0
+    for (const line of lines) size += Buffer.byteLength(line, 'utf8') + 1
+    if (this.bytes + size > this.maxBytes) {
+      this.truncated = true
+      return false
+    }
+    this.lines.push(...lines)
+    this.bytes += size
+    return true
+  }
+
+  /** Joins the accepted lines, ending with the truncation notice when a push was refused. */
+  join(): string {
+    return this.truncated ? [...this.lines, TRUNCATION_NOTICE].join('\n') : this.lines.join('\n')
+  }
+}

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

@@ -446,7 +446,13 @@ export async function sweepStaleMemberObservations(now: Date): Promise<StaleMemb
     ${MEMBER_OBSERVATION_STALE_AFTER_HOURS} * INTERVAL '1 hour',
     2 * ${knowledgeConnector.syncIntervalMinutes} * INTERVAL '1 minute'
   )`
-  const cutoff = sql`${sql.param(now, knowledgeConnectorMember.lastStartedAt)} - ${staleWindow}`
+  /**
+   * The bound instant is cast: in `$now - GREATEST(...)` Postgres cannot see a
+   * timestamp on either side and resolves the subtraction as interval
+   * arithmetic, which makes the cutoff an interval and every comparison below
+   * fail with "operator does not exist: timestamp > interval".
+   */
+  const cutoff = sql`${sql.param(now, knowledgeConnectorMember.lastStartedAt)}::timestamp - ${staleWindow}`
   const staleMembers = await db
     .select({
       id: knowledgeConnectorMember.id,

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

@@ -26,6 +26,7 @@ import {
   memberFailureBackoffMs,
   memberNextAttemptAt,
   nextMemberSyncTime,
+  persistedDocumentsByObserver,
   shouldListFully,
 } from '@/lib/knowledge/connectors/member-sync-engine'
 import {
@@ -272,6 +273,26 @@ describe('member sync engine decisions', () => {
       expect(second.retainedBytes).toBeGreaterThan(first.retainedBytes)
     })
 
+    it('grants a persisted batch to every member who listed each document, as it lands', () => {
+      const union = new Map()
+      admitMemberListing(union, 'm-1', [doc('a'), doc('b')], 'c-1', 0)
+      admitMemberListing(union, 'm-2', [doc('a'), doc('c')], 'c-1', 0)
+
+      const byMember = persistedDocumentsByObserver(
+        [
+          { externalId: 'a', documentId: 'd-a' },
+          { externalId: 'b', documentId: 'd-b' },
+          { externalId: 'zzz', documentId: 'd-z' },
+        ],
+        union
+      )
+
+      expect([...byMember.entries()]).toEqual([
+        ['m-1', ['d-a', 'd-b']],
+        ['m-2', ['d-a']],
+      ])
+    })
+
     it('counts a member once per external id even when their listing repeats it', () => {
       const union = new Map()
       const admitted = admitMemberListing(union, 'm-1', [doc('a'), doc('a')], 'c-1', 0)