fix(gmail): open search results in the indexed mailbox (#7910)

Author: waleedlatif1

apps/sim/connectors/gmail/company-crawl.test.ts

@@ -64,7 +64,11 @@ function providerResponse(url: string, init?: RequestInit): Response {
   const parsed = new URL(url)
   const token = new Headers(init?.headers).get('Authorization')
   const mailbox = token?.includes(BOB.email) ? 'Bob' : 'Alice'
-  if (parsed.pathname.endsWith('/profile')) return Response.json({ emailAddress: ALICE.email })
+  if (parsed.pathname.endsWith('/profile'))
+    return Response.json({
+      emailAddress: mailbox === 'Bob' ? BOB.email : ALICE.email,
+      historyId: '100',
+    })
   if (parsed.pathname.endsWith('/labels')) {
     return Response.json({ labels: [{ id: 'Label_7', name: `${mailbox} label` }] })
   }
@@ -234,6 +238,7 @@ describe('company-wide Gmail indexing', () => {
     const context = centralContext()
     const first = await gmailConnector.listDocuments('directory-token', CONFIG, undefined, context)
     const alice = first.documents[0]
+    expect(new URL(alice.sourceUrl!).searchParams.get('Email')).toBe(ALICE.email)
     const aliceBody = await gmailConnector.getDocument(
       'directory-token',
       CONFIG,
@@ -247,6 +252,7 @@ describe('company-wide Gmail indexing', () => {
       context
     )
     const bob = second.documents[0]
+    expect(new URL(bob.sourceUrl!).searchParams.get('Email')).toBe(BOB.email)
     const bobBody = await gmailConnector.getDocument(
       'directory-token',
       CONFIG,
@@ -322,7 +328,10 @@ describe('company-wide Gmail indexing', () => {
       resumedContext
     )
     expect(resumed.documents).toEqual(first.documents)
-    expect(new URL(fetchProvider.mock.calls[1][0]).searchParams.get('q')).toBe(firstQuery)
+    const listings = fetchProvider.mock.calls.filter(([url]) =>
+      new URL(url).pathname.endsWith('/threads')
+    )
+    expect(new URL(listings[1][0]).searchParams.get('q')).toBe(firstQuery)
     const body = await gmailConnector.getDocument(
       'directory-token',
       CONFIG,

apps/sim/connectors/gmail/gmail.test.ts

@@ -17,6 +17,12 @@ vi.mock('@/lib/knowledge/documents/secure-fetch.server', () => ({
       ? options.fetcher(url, init, mockFetchWithRetry)
       : mockFetchWithRetry(url, init),
 }))
+vi.mock('@/connectors/gmail/mailbox', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@/connectors/gmail/mailbox')>()),
+  getGmailMailboxEmail: vi.fn(async (token: string) =>
+    token === 'bob-token' ? 'bob@example.com' : 'alice@example.com'
+  ),
+}))
 vi.mock('@/components/icons', () => ({ GmailIcon: () => null }))
 vi.mock('@/lib/knowledge/documents/service', () => ({
   isTriggerAvailable: () => false,
@@ -789,7 +795,8 @@ describe('Gmail Search member isolation', () => {
       externalId: 'member:alice:thread-1',
       contentHash: 'gmail:thread-1:10:body-v2',
       contentDeferred: false,
-      sourceUrl: 'https://mail.google.com/mail/u/0/#all/thread-1',
+      sourceUrl:
+        'https://accounts.google.com/AccountChooser?Email=alice%40example.com&continue=https%3A%2F%2Fmail.google.com%2Fmail%2F%3Fauthuser%3Dalice%2540example.com%23all%2Fthread-1',
     })
     expect(document?.content).toContain('Private mailbox content')
     expect(mockFetchWithRetry.mock.calls[0][0]).toContain('/threads/thread-1?format=full')
@@ -1221,7 +1228,8 @@ describe('Gmail change feed', () => {
     mockFetchWithRetry.mockImplementation(async (url: string) => {
       const parsed = new URL(url)
       requests.push(parsed)
-      if (parsed.pathname.endsWith('/profile')) return Response.json({ historyId: '500' })
+      if (parsed.pathname.endsWith('/profile'))
+        return Response.json({ emailAddress: 'alice@example.com', historyId: '500' })
       if (parsed.pathname.endsWith('/labels')) return Response.json({ labels })
       if (parsed.pathname.endsWith('/history')) {
         return Response.json(pages[historyCall++] ?? historyPage([]))

apps/sim/connectors/gmail/gmail.ts

@@ -4,6 +4,7 @@ import { isPlainRecord } from '@sim/utils/object'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { isPayloadSizeLimitError, readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits'
 import { VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils'
+import { getGmailMailboxEmail, getGmailProfile, gmailThreadUrl } from '@/connectors/gmail/mailbox'
 import { DEFAULT_MAX_THREADS, gmailConnectorMeta } from '@/connectors/gmail/meta'
 import { fetchGoogleApiWithRetry, GoogleApiError } from '@/connectors/google-workspace/api-errors'
 import {
@@ -621,34 +622,27 @@ async function resolveLabelNames(
  * Creates a lightweight document stub from a thread list entry.
  * Uses metadata-based contentHash for change detection without downloading content.
  */
-function threadToStub(
+async function threadToStub(
   thread: GmailThread,
+  accessToken: string,
   syncContext?: Record<string, unknown>
-): ExternalDocument {
+): Promise<ExternalDocument> {
+  const mailboxEmail = await getGmailMailboxEmail(accessToken, syncContext)
   return {
     externalId: memberDocumentId(thread.id, syncContext),
     title: thread.snippet || 'Untitled Thread',
     content: '',
     contentDeferred: true,
     estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
     mimeType: 'text/plain',
-    sourceUrl: threadUrl(thread.id),
+    sourceUrl: gmailThreadUrl(thread.id, mailboxEmail),
     /** Rehydrate older rows that omitted separately stored message bodies. */
     contentHash: `gmail:${thread.id}:${thread.historyId}:body-v2`,
     skippedRetryPolicy: 'source-change',
     metadata: {},
   }
 }
 
-/**
- * Deep link to a thread. `#all` is used rather than `#inbox` because a synced
- * thread may be archived or live only under a user label, where an `#inbox`
- * fragment resolves to nothing.
- */
-function threadUrl(threadId: string): string {
-  return `https://mail.google.com/mail/u/0/#all/${threadId}`
-}
-
 /** A feed position: the mailbox history id the next read starts from, mid-page when paging. */
 interface GmailChangeCursor {
   historyId: string
@@ -780,22 +774,7 @@ const gmailMailboxConnector: ConnectorConfig = {
     if (syncContext?.mirrorsSourceAcls === true) {
       throw new Error('Company-wide Gmail indexing uses complete mailbox listings')
     }
-    const response = await fetchGoogleApiWithRetry(
-      'gmail.users.getProfile',
-      `${GMAIL_API_BASE}/profile?fields=historyId`,
-      {
-        method: 'GET',
-        headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' },
-      }
-    )
-    const data: unknown = await response.json()
-    if (
-      !isPlainRecord(data) ||
-      typeof data.historyId !== 'string' ||
-      !/^\d+$/.test(data.historyId)
-    ) {
-      throw new Error('Gmail returned malformed profile metadata')
-    }
+    const data = await getGmailProfile(accessToken, syncContext)
     return JSON.stringify({ historyId: data.historyId })
   },
 
@@ -813,7 +792,7 @@ const gmailMailboxConnector: ConnectorConfig = {
     accessToken: string,
     sourceConfig: Record<string, unknown>,
     cursor: string,
-    syncContext?: Record<string, unknown>
+    syncContext: Record<string, unknown> = {}
   ): Promise<ExternalChangeList> => {
     if (syncContext?.mirrorsSourceAcls === true) {
       throw new Error('Company-wide Gmail indexing uses complete mailbox listings')
@@ -854,7 +833,11 @@ const gmailMailboxConnector: ConnectorConfig = {
         const externalId = memberDocumentId(threadId, syncContext)
         const thread = await fetchThread(accessToken, threadId, 'metadata')
         if (!thread || !threadInScope(thread, scope)) return { kind: 'removed', externalId }
-        return { kind: 'upsert', externalId, document: threadToStub(thread, syncContext) }
+        return {
+          kind: 'upsert',
+          externalId,
+          document: await threadToStub(thread, accessToken, syncContext),
+        }
       }
     )
 
@@ -873,7 +856,7 @@ const gmailMailboxConnector: ConnectorConfig = {
     accessToken: string,
     sourceConfig: Record<string, unknown>,
     cursor?: string,
-    syncContext?: Record<string, unknown>
+    syncContext: Record<string, unknown> = {}
   ): Promise<ExternalDocumentList> => {
     const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined
     signal?.throwIfAborted()
@@ -962,7 +945,7 @@ const gmailMailboxConnector: ConnectorConfig = {
       const metadata = thread.historyId
         ? thread
         : await fetchThread(accessToken, thread.id, 'minimal', signal)
-      return metadata ? threadToStub(metadata, syncContext) : null
+      return metadata ? threadToStub(metadata, accessToken, syncContext) : null
     })
     const documents = stubs.filter((stub): stub is ExternalDocument => stub !== null)
 
@@ -1002,7 +985,7 @@ const gmailMailboxConnector: ConnectorConfig = {
     accessToken: string,
     _sourceConfig: Record<string, unknown>,
     externalId: string,
-    syncContext?: Record<string, unknown>
+    syncContext: Record<string, unknown> = {}
   ): Promise<ExternalDocument | null> => {
     const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined
     signal?.throwIfAborted()
@@ -1028,7 +1011,7 @@ const gmailMailboxConnector: ConnectorConfig = {
         }
         return {
           ...markSkipped(
-            threadToStub(after, syncContext),
+            await threadToStub(after, accessToken, syncContext),
             sizeLimitSkipReason(MAX_THREAD_RESPONSE_BYTES)
           ),
           skippedExistingDisposition: 'replace',
@@ -1043,7 +1026,10 @@ const gmailMailboxConnector: ConnectorConfig = {
     } catch (error) {
       if (error instanceof ConnectorFileTooLargeError) {
         return {
-          ...markSkipped(threadToStub(thread, syncContext), sizeLimitSkipReason(error.limitBytes)),
+          ...markSkipped(
+            await threadToStub(thread, accessToken, syncContext),
+            sizeLimitSkipReason(error.limitBytes)
+          ),
           skippedExistingDisposition: 'replace',
         }
       }
@@ -1056,7 +1042,7 @@ const gmailMailboxConnector: ConnectorConfig = {
     metadata.labels = await resolveLabelNames(accessToken, labelIds, syncContext)
 
     return {
-      ...threadToStub(thread, syncContext),
+      ...(await threadToStub(thread, accessToken, syncContext)),
       title: subject,
       content,
       contentDeferred: false,

apps/sim/connectors/gmail/mailbox.test.ts

@@ -0,0 +1,112 @@
+/** @vitest-environment node */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { getGmailMailboxEmail, getGmailProfile, gmailThreadUrl } from '@/connectors/gmail/mailbox'
+
+const { transport } = vi.hoisted(() => ({ transport: vi.fn<typeof fetch>() }))
+vi.mock('@/lib/knowledge/documents/secure-fetch.server', () => ({
+  fetchWithRetry: (
+    url: string,
+    init: RequestInit,
+    options: {
+      fetcher: (url: string, init: RequestInit, transport: typeof fetch) => Promise<Response>
+    }
+  ) => options.fetcher(url, init, transport),
+}))
+
+beforeEach(() => {
+  transport.mockReset()
+  transport.mockResolvedValue(
+    Response.json({ emailAddress: 'Alice+work@example.com', historyId: '123' })
+  )
+})
+
+describe('Gmail mailbox identity', () => {
+  it('selects the authenticated mailbox before an archived-thread fragment', async () => {
+    const email = await getGmailMailboxEmail('mailbox-token', {})
+    const url = new URL(gmailThreadUrl('19a3f0123456789', email))
+    expect(url.origin).toBe('https://accounts.google.com')
+    expect(url.pathname).toBe('/AccountChooser')
+    expect(url.searchParams.get('Email')).toBe('alice+work@example.com')
+    expect(url.hash).toBe('')
+    const destination = new URL(url.searchParams.get('continue')!)
+    expect(destination.origin).toBe('https://mail.google.com')
+    expect(destination.pathname).toBe('/mail/')
+    expect(destination.searchParams.get('authuser')).toBe('alice+work@example.com')
+    expect(destination.hash).toBe('#all/19a3f0123456789')
+    expect(transport).toHaveBeenCalledWith(
+      'https://gmail.googleapis.com/gmail/v1/users/me/profile?fields=emailAddress,historyId',
+      expect.objectContaining({
+        headers: expect.objectContaining({ Authorization: 'Bearer mailbox-token' }),
+      })
+    )
+  })
+
+  it('deduplicates concurrent lookups and reuses the profile read for the history watermark', async () => {
+    const context = {}
+    expect(
+      await Promise.all(Array.from({ length: 100 }, () => getGmailMailboxEmail('token', context)))
+    ).toEqual(Array(100).fill('alice+work@example.com'))
+    expect(transport).toHaveBeenCalledTimes(1)
+    transport.mockResolvedValue(
+      Response.json({ emailAddress: 'alice+work@example.com', historyId: '456' })
+    )
+    expect((await getGmailProfile('token', context)).historyId).toBe('456')
+    expect(await getGmailMailboxEmail('token', context)).toBe('alice+work@example.com')
+    expect(transport).toHaveBeenCalledTimes(2)
+  })
+
+  it('seeds mailbox identity from a fresh watermark without a second request', async () => {
+    const context = {}
+    await getGmailProfile('token', context)
+    await getGmailMailboxEmail('token', context)
+    expect(transport).toHaveBeenCalledTimes(1)
+  })
+
+  it('isolates credentials even if the same context is mistakenly reused', async () => {
+    const context = {}
+    await getGmailMailboxEmail('alice-token', context)
+    transport.mockImplementation(async () =>
+      Response.json({ emailAddress: 'bob@example.com', historyId: '321' })
+    )
+    expect(await getGmailMailboxEmail('bob-token', context)).toBe('bob@example.com')
+    await getGmailMailboxEmail('bob-token', {})
+    expect(transport).toHaveBeenCalledTimes(3)
+  })
+
+  it.each([
+    {},
+    { emailAddress: '', historyId: '1' },
+    { emailAddress: 'not-an-email', historyId: '1' },
+    { emailAddress: 'alice@example.com', historyId: 123 },
+    { emailAddress: 'alice@example.com', historyId: 'invalid' },
+  ])('rejects malformed profiles instead of falling back to account zero: %j', async (body) => {
+    transport.mockResolvedValue(Response.json(body))
+    await expect(getGmailMailboxEmail('token', {})).rejects.toThrow('malformed profile')
+  })
+
+  it('does not cache a failed lookup', async () => {
+    transport.mockResolvedValueOnce(Response.json({}))
+    const context = {}
+    await expect(getGmailMailboxEmail('token', context)).rejects.toThrow()
+    expect(await getGmailMailboxEmail('token', context)).toBe('alice+work@example.com')
+  })
+
+  it('preserves provider authentication errors and cancellation', async () => {
+    transport.mockResolvedValueOnce(new Response(null, { status: 401 }))
+    await expect(getGmailMailboxEmail('invalid', {})).rejects.toMatchObject({ status: 401 })
+    const controller = new AbortController()
+    controller.abort()
+    await expect(getGmailProfile('token', { signal: controller.signal })).rejects.toThrow()
+    expect(transport).toHaveBeenCalledTimes(1)
+  })
+
+  it('bounds profile responses', async () => {
+    transport.mockResolvedValue(Response.json({ padding: 'x'.repeat(17 * 1024) }))
+    await expect(getGmailMailboxEmail('token', {})).rejects.toThrow()
+  })
+
+  it('encodes fragment delimiters rather than allowing them to change the link target', () => {
+    const chooser = new URL(gmailThreadUrl('thread/?#id', 'alice@example.com'))
+    expect(new URL(chooser.searchParams.get('continue')!).hash).toBe('#all/thread%2F%3F%23id')
+  })
+})