fix(connectors): keep the newest messages when a chat transcript reaches the size limit

Author: waleedlatif1

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

@@ -328,24 +328,27 @@ function formatSpaceContent(
   space: Space,
   messages: ChatMessage[]
 ): { content: string; messageCount: number } {
-  const parts = new BoundedLines()
-  parts.push(`Space: ${spaceTitle(space)}`)
+  /** The newest messages survive when the window does not fit; the header always does. */
+  const parts = new BoundedLines(CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, 'last')
+  parts.pin(`Space: ${spaceTitle(space)}`)
   const description = space.spaceDetails?.description?.trim()
-  if (description) parts.push(`Description: ${description}`)
+  if (description) parts.pin(`Description: ${description}`)
   const guidelines = space.spaceDetails?.guidelines?.trim()
-  if (guidelines) parts.push(`Guidelines: ${guidelines}`)
+  if (guidelines) parts.pin(`Guidelines: ${guidelines}`)
 
-  let messageCount = 0
+  let headed = false
   for (const message of messages) {
     const text = message.text?.trim() || message.fallbackText?.trim()
     if (!text) continue
-    if (messageCount === 0) parts.push('', '--- Messages ---')
+    if (!headed) {
+      parts.pin('', '--- Messages ---')
+      headed = true
+    }
     const timestamp = message.createTime ?? ''
-    if (!parts.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`)) break
-    messageCount += 1
+    parts.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`)
   }
 
-  return { content: parts.join(), messageCount }
+  return { content: parts.join(), messageCount: parts.count }
 }
 
 export const googleChatConnector: ConnectorConfig = {

apps/sim/connectors/slack/slack.ts

@@ -385,13 +385,17 @@ function walkBlockText(node: unknown, out: string[]): void {
  * Each entry: "[ISO timestamp] username: message text" (text may span lines
  * when the message has rich attachment/block content).
  */
+/**
+ * Appends the messages to the transcript oldest first. The transcript keeps
+ * its newest messages when the window does not fit, so a message is skipped
+ * only when it cannot fit on its own.
+ */
 async function appendMessages(
   accessToken: string,
   lines: BoundedLines,
   messages: SlackMessage[],
   syncContext?: Record<string, unknown>
-): Promise<number> {
-  let appended = 0
+): Promise<void> {
   /** Slack returns newest first; the transcript reads oldest first. */
   const chronological = [...messages].reverse()
 
@@ -412,11 +416,8 @@ async function appendMessages(
       ? await resolveUserName(accessToken, msg.user, syncContext)
       : msg.username || 'unknown'
 
-    if (!lines.push(`[${timestamp}] ${userName}: ${content}`)) break
-    appended += 1
+    lines.push(`[${timestamp}] ${userName}: ${content}`)
   }
-
-  return appended
 }
 
 /**
@@ -601,14 +602,15 @@ async function buildSlackChannelDocument(
     maxMessages
   )
 
-  const lines = new BoundedLines()
-  lines.push(`Channel: #${channel.name}`)
+  const lines = new BoundedLines(CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, 'last')
+  lines.pin(`Channel: #${channel.name}`)
   const topic = channel.topic?.value?.trim()
-  if (topic) lines.push(`Topic: ${topic}`)
+  if (topic) lines.pin(`Topic: ${topic}`)
   const purpose = channel.purpose?.value?.trim()
-  if (purpose) lines.push(`Purpose: ${purpose}`)
-  lines.push('')
-  const messageCount = await appendMessages(accessToken, lines, messages, syncContext)
+  if (purpose) lines.pin(`Purpose: ${purpose}`)
+  lines.pin('')
+  await appendMessages(accessToken, lines, messages, syncContext)
+  const messageCount = lines.count
 
   /**
    * Edit/thread fingerprint: max(edited.ts) and max(latest_reply) across the

apps/sim/connectors/utils.test.ts

@@ -1662,4 +1662,37 @@ describe('BoundedLines', () => {
     expect(lines.push('éé')).toBe(true)
     expect(lines.push('é')).toBe(false)
   })
+
+  describe('keeping the last records', () => {
+    it('lets the oldest records go so the newest fit, under a header that stays', () => {
+      const lines = new BoundedLines(24, 'last')
+      lines.pin('# room')
+      expect(lines.push('one')).toBe(true)
+      expect(lines.push('two')).toBe(true)
+      expect(lines.push('three')).toBe(true)
+      expect(lines.push('four')).toBe(true)
+      expect(lines.count).toBe(3)
+      expect(lines.join()).toBe(
+        '# room\n[Truncated: earlier text was left out to fit the size limit]\ntwo\nthree\nfour'
+      )
+    })
+
+    it('refuses only a record that cannot fit on its own and carries on', () => {
+      const lines = new BoundedLines(12, 'last')
+      expect(lines.push('a very long record')).toBe(false)
+      expect(lines.push('short')).toBe(true)
+      expect(lines.push('next')).toBe(true)
+      expect(lines.count).toBe(2)
+      expect(lines.join()).toBe(
+        '[Truncated: earlier text was left out to fit the size limit]\nshort\nnext'
+      )
+    })
+
+    it('joins the header and records plainly when everything fits', () => {
+      const lines = new BoundedLines(64, 'last')
+      lines.pin('# room', '')
+      lines.push('hello')
+      expect(lines.join()).toBe('# room\n\nhello')
+    })
+  })
 })

apps/sim/connectors/utils.ts

@@ -806,39 +806,96 @@ export function isSkippableMicrosoftGraphFolderError(
  */
 export const CONNECTOR_TEXT_DOCUMENT_MAX_BYTES = 12 * 1024 * 1024
 
-const TRUNCATION_NOTICE = '[Truncated: the indexed text reached the size limit]'
+const TRAILING_TRUNCATION_NOTICE = '[Truncated: the indexed text reached the size limit]'
+const LEADING_TRUNCATION_NOTICE = '[Truncated: earlier text was left out to fit 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.
+ * Which end of the stream survives when it does not fit: `first` keeps what
+ * was pushed first and refuses the rest (a mail thread, whose root message is
+ * the context), `last` keeps what was pushed last and lets older records go
+ * (a chat transcript, whose newest messages are the ones people search for).
+ */
+export type BoundedLinesKeep = 'first' | 'last'
+
+interface BoundedRecord {
+  lines: string[]
+  bytes: number
+}
+
+function byteSize(lines: readonly string[]): number {
+  let size = 0
+  for (const line of lines) size += Buffer.byteLength(line, 'utf8') + 1
+  return size
+}
+
+/**
+ * Accumulates newline-joined text under a byte ceiling. A record is kept
+ * whole or not at all, so a truncated document never ends mid-message, and
+ * the output carries a notice where something was left out.
  */
 export class BoundedLines {
-  private readonly lines: string[] = []
-  private bytes = 0
+  private readonly pinned: string[] = []
+  private readonly records: BoundedRecord[] = []
+  private pinnedBytes = 0
+  private recordBytes = 0
   private truncated = false
 
-  constructor(private readonly maxBytes = CONNECTOR_TEXT_DOCUMENT_MAX_BYTES) {}
+  constructor(
+    private readonly maxBytes = CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
+    private readonly keep: BoundedLinesKeep = 'first'
+  ) {}
+
+  /** Lines that open the document and are never let go, such as its header; counted against the ceiling. */
+  pin(...lines: string[]): void {
+    this.pinned.push(...lines)
+    this.pinnedBytes += byteSize(lines)
+  }
+
+  /** Records currently kept. */
+  get count(): number {
+    return this.records.length
+  }
 
   /**
-   * Appends the lines together when they fit; otherwise marks the document
-   * truncated, appends nothing, and returns false so the caller stops.
+   * Appends the lines as one record and returns whether it was kept. Keeping
+   * the first, a record that does not fit is refused, and so is every later
+   * one, so a caller can stop. Keeping the last, a record is refused only when
+   * it cannot fit on its own, and appending it lets the oldest records go
+   * until the rest fits, so a caller carries on.
    */
   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) {
+    const bytes = byteSize(lines)
+    if (this.keep === 'first') {
+      if (this.truncated) return false
+      if (this.pinnedBytes + this.recordBytes + bytes > this.maxBytes) {
+        this.truncated = true
+        return false
+      }
+      this.records.push({ lines, bytes })
+      this.recordBytes += bytes
+      return true
+    }
+    if (this.pinnedBytes + bytes > this.maxBytes) {
       this.truncated = true
       return false
     }
-    this.lines.push(...lines)
-    this.bytes += size
+    this.records.push({ lines, bytes })
+    this.recordBytes += bytes
+    while (this.pinnedBytes + this.recordBytes > this.maxBytes) {
+      const oldest = this.records.shift()
+      if (!oldest) break
+      this.recordBytes -= oldest.bytes
+      this.truncated = true
+    }
     return true
   }
 
-  /** Joins the accepted lines, ending with the truncation notice when a push was refused. */
+  /** Joins the kept lines, with the truncation notice where records were left out. */
   join(): string {
-    return this.truncated ? [...this.lines, TRUNCATION_NOTICE].join('\n') : this.lines.join('\n')
+    const body = this.records.flatMap((record) => record.lines)
+    if (!this.truncated) return [...this.pinned, ...body].join('\n')
+    return this.keep === 'first'
+      ? [...this.pinned, ...body, TRAILING_TRUNCATION_NOTICE].join('\n')
+      : [...this.pinned, LEADING_TRUNCATION_NOTICE, ...body].join('\n')
   }
 }