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

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,
@@ -398,13 +396,6 @@ async function resolveLabelNames(
     .filter((name) => !name.startsWith('CATEGORY_') && name !== 'UNREAD')
 }
 
-/**
- * What a hydrated thread retains: message bodies as text, never attachments,
- * so a generous ceiling for a long thread is still small. Lets a batch of
- * threads hydrate together instead of one at a time.
- */
-const ESTIMATED_THREAD_BYTES = 256 * 1024
-
 /**
  * Creates a lightweight document stub from a thread list entry.
  * Uses metadata-based contentHash for change detection without downloading content.
@@ -419,7 +410,7 @@ function threadToStub(thread: {
     title: thread.snippet || 'Untitled Thread',
     content: '',
     contentDeferred: true,
-    estimatedBytes: ESTIMATED_THREAD_BYTES,
+    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,13 +9,10 @@ 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')
 
-/** A hydrated space retains a bounded window of message text; lets spaces hydrate in batches. */
-const ESTIMATED_SPACE_BYTES = 1024 * 1024
-
 const CHAT_API_BASE = 'https://chat.googleapis.com/v1'
 const MS_PER_DAY = 24 * 60 * 60 * 1000
 
@@ -221,7 +218,7 @@ function spaceToStub(
     title: spaceTitle(space),
     content: '',
     contentDeferred: true,
-    estimatedBytes: ESTIMATED_SPACE_BYTES,
+    estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
     mimeType: 'text/plain',
     sourceUrl: space.spaceUri,
     contentHash: buildContentHash(space, maxMessages, lookbackDays, syncContext),
@@ -328,27 +325,26 @@ function senderLabel(sender: ChatUser | undefined): string {
  * instead, which already handles size caps, OCR, and format parsing.
  */
 function formatSpaceContent(space: Space, messages: ChatMessage[]): string {
-  const parts: string[] = [`Space: ${spaceTitle(space)}`]
+  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 headed = false
   for (const message of messages) {
     const text = message.text?.trim() || message.fallbackText?.trim()
     if (!text) continue
+    if (!headed) {
+      parts.push('', '--- Messages ---')
+      headed = true
+    }
     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
   }
 
-  return parts.join('\n')
+  return parts.join()
 }
 
 /** Number of messages that actually contributed text to the transcript. */

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,
@@ -13,9 +15,6 @@ import {
 
 const logger = createLogger('OutlookConnector')
 
-/** A hydrated conversation retains message bodies as text, never attachments; lets threads hydrate in batches. */
-const ESTIMATED_CONVERSATION_BYTES = 256 * 1024
-
 const GRAPH_API_BASE = 'https://graph.microsoft.com/v1.0/me'
 const MESSAGES_PER_PAGE = 50
 /**
@@ -566,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>()
@@ -808,7 +803,7 @@ export const outlookConnector: ConnectorConfig = {
         title: subject,
         content: '',
         contentDeferred: true,
-        estimatedBytes: ESTIMATED_CONVERSATION_BYTES,
+        estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
         mimeType: 'text/plain',
         sourceUrl,
         contentHash: `outlook:${convId}:${lastDate}`,

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')
+  }
+}