fix(google-chat): avoid unchanged transcript reindexing (#8925)

* fix(google-chat): avoid unchanged transcript reindexing

* fix(google-chat): reject rounded fractional settings
This commit is contained in:
Waleed
2026-10-12 10:59:31 -07:00
committed by GitHub
parent 6511d43cb4
commit 0951bde7bf
5 changed files with 307 additions and 36 deletions
@@ -1,5 +1,6 @@
import { createServer } from 'node:http'
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { classifyExternalDoc, contentHashMatch } from '@/lib/knowledge/connectors/sync-primitives'
import { googleChatConnector } from '@/connectors/google-chat/google-chat'
const SPACE = {
@@ -16,6 +17,8 @@ let messageReads = 0
let requestCount = 0
let firstParameters = ''
let now = Date.parse('2026-10-02T12:00:00Z')
let activity = SPACE.lastActiveTime
let body = 'New decision: review on Friday.'
const server = createServer((request, response) => {
requestCount++
@@ -29,7 +32,7 @@ const server = createServer((request, response) => {
return
}
if (url.pathname === '/v1/spaces') {
send({ spaces: [SPACE] })
send({ spaces: [{ ...SPACE, lastActiveTime: activity }] })
return
}
if (url.pathname === `/v1/${SPACE.name}`) {
@@ -81,6 +84,28 @@ const server = createServer((request, response) => {
})
return
}
if (mode === 'projected') {
const fields = url.searchParams.get('fields') ?? '*'
if (fields === '*' || fields.includes('cards')) {
send({ cards: 'x'.repeat(33 * 1024 * 1024) })
return
}
if (
![
'nextPageToken',
'messages',
'createTime',
'sender',
'name',
'displayName',
'text',
'fallbackText',
].every((field) => fields.includes(field))
) {
send({ error: 'Transcript or continuation field omitted' }, 400)
return
}
}
if (mode === 'empty-page' && !token) {
send({ nextPageToken: 'page-2' })
return
@@ -89,14 +114,14 @@ const server = createServer((request, response) => {
const cutoff = url.searchParams.get('filter')?.match(/"([^"]+)"/)?.[1]
send({
messages:
cutoff && time <= cutoff
cutoff && Date.parse(time) <= Date.parse(cutoff)
? []
: [
{
name: `${SPACE.name}/messages/${token ? 'older' : 'newer'}`,
sender: { name: 'users/fixture-author' },
createTime: time,
text: token ? 'Older decision: launch is paused.' : 'New decision: review on Friday.',
createTime: mode === 'unknown-time' ? undefined : time,
text: token ? 'Older decision: launch is paused.' : body,
},
],
nextPageToken:
@@ -123,6 +148,8 @@ beforeEach(() => {
requestCount = 0
firstParameters = ''
now = Date.parse('2026-10-02T12:00:00Z')
activity = SPACE.lastActiveTime
body = 'New decision: review on Friday.'
})
afterAll(async () => {
@@ -184,15 +211,176 @@ describe('Google Chat connector over real HTTP', () => {
const first = await googleChatConnector.listDocuments(TOKEN, config, undefined, firstContext)
now += 60_000
const initial = await read(config, firstContext)
if (!initial) throw new Error('Expected a space document')
expect(initial?.content).toContain('New decision:')
expect(initial?.contentHash).toBe(first.documents[0].contentHash)
expect(
contentHashMatch(
first.documents[0].contentHash,
initial.contentHash,
googleChatConnector.matchContentHash
)
).toBe('current')
now += 3 * 24 * 60 * 60 * 1000
const nextContext = {}
const next = await googleChatConnector.listDocuments(TOKEN, config, undefined, nextContext)
expect(next.documents[0].contentHash).not.toBe(first.documents[0].contentHash)
const expired = await read(config, nextContext)
if (!expired) throw new Error('Expected a space document')
expect(expired?.content).not.toContain('New decision:')
expect(expired?.contentHash).toBe(next.documents[0].contentHash)
expect(
contentHashMatch(
next.documents[0].contentHash,
expired.contentHash,
googleChatConnector.matchContentHash
)
).toBe('current')
})
it('skips unchanged rolling-window spaces until the exact next message expiry', async () => {
mode = 'window'
vi.spyOn(Date, 'now').mockImplementation(() => now)
const config = { lookbackDays: '2' }
const context = {}
await googleChatConnector.listDocuments(TOKEN, config, undefined, context)
const initial = await read(config, context)
if (!initial) throw new Error('Expected a space document')
const existing = { id: 'stored', contentHash: initial.contentHash, storageKey: 'indexed' }
const classify = async () => {
const page = await googleChatConnector.listDocuments(TOKEN, config, undefined, {})
return classifyExternalDoc(
page.documents[0],
existing,
false,
googleChatConnector.matchContentHash
).type
}
now += 1
expect(await classify()).toBe('unchanged')
now = Date.parse('2026-10-03T11:59:59.999Z')
expect(await classify()).toBe('unchanged')
expect(messageReads).toBe(1)
now = Date.parse('2026-10-02T11:59:59.999Z')
expect(await classify()).toBe('update')
now = Date.parse('2026-10-03T11:59:59.999Z')
now += 1
expect(await classify()).toBe('update')
const expired = await read(config, {})
expect(expired?.content).not.toContain('New decision:')
expect(messageReads).toBe(2)
})
it('keeps empty windows cheap but refreshes changed settings, full syncs, and legacy hashes', async () => {
mode = 'window'
now = Date.parse('2026-10-04T12:00:00Z')
vi.spyOn(Date, 'now').mockImplementation(() => now)
const config = { lookbackDays: '2' }
const initial = await read(config, {})
if (!initial) throw new Error('Expected a space document')
const existing = { id: 'stored', contentHash: initial.contentHash, storageKey: 'indexed' }
now += 24 * 60 * 60 * 1000
const page = await googleChatConnector.listDocuments(TOKEN, config, undefined, {})
const stub = page.documents[0]
expect(
classifyExternalDoc(stub, existing, false, googleChatConnector.matchContentHash).type
).toBe('unchanged')
expect(
classifyExternalDoc(stub, existing, true, googleChatConnector.matchContentHash).type
).toBe('update')
expect(
classifyExternalDoc(
stub,
{ ...existing, contentHash: 'gchat:v2:legacy' },
false,
googleChatConnector.matchContentHash
).type
).toBe('update')
const changed = await googleChatConnector.listDocuments(
TOKEN,
{ ...config, maxMessages: '100' },
undefined,
{}
)
expect(
classifyExternalDoc(
changed.documents[0],
existing,
false,
googleChatConnector.matchContentHash
).type
).toBe('update')
})
it('rereads a lookback window when the provider omits creation timestamps', async () => {
mode = 'unknown-time'
vi.spyOn(Date, 'now').mockImplementation(() => now)
const config = { lookbackDays: '90' }
const initial = await read(config, {})
if (!initial) throw new Error('Expected a space document')
now += 60_000
const page = await googleChatConnector.listDocuments(TOKEN, config, undefined, {})
const existing = { id: 'stored', contentHash: initial.contentHash, storageKey: 'indexed' }
expect(
classifyExternalDoc(page.documents[0], existing, false, googleChatConnector.matchContentHash)
.type
).toBe('update')
})
it('advances source metadata without replacing identical text but still indexes changed text', async () => {
mode = 'window'
vi.spyOn(Date, 'now').mockImplementation(() => now)
const load = async () => {
const context = {}
await googleChatConnector.listDocuments(TOKEN, {}, undefined, context)
const doc = await read({}, context)
if (!doc) throw new Error('Expected a space document')
return doc
}
const initial = await load()
activity = '2026-10-02T12:00:00Z'
const page = await googleChatConnector.listDocuments(TOKEN, {}, undefined, {})
expect(
classifyExternalDoc(
page.documents[0],
{ id: 'stored', contentHash: initial.contentHash, storageKey: 'indexed' },
false,
googleChatConnector.matchContentHash
).type
).toBe('update')
const refreshed = await load()
expect(
contentHashMatch(
refreshed.contentHash,
initial.contentHash,
googleChatConnector.matchContentHash
)
).toBe('equivalent')
body = 'Changed decision: launch is approved.'
const changed = await load()
expect(
contentHashMatch(
changed.contentHash,
refreshed.contentHash,
googleChatConnector.matchContentHash
)
).toBe('stale')
})
it.each(['100.0', '1e2'])(
'keeps saved whole-number strings such as %s usable through validation and hydration',
async (maxSpaces) => {
vi.spyOn(Date, 'now').mockImplementation(() => now)
const config = { maxSpaces, maxMessages: '3.00', lookbackDays: '90.0' }
expect(await googleChatConnector.validateConfig(TOKEN, config)).toEqual({ valid: true })
const context = {}
await googleChatConnector.listDocuments(TOKEN, config, undefined, context)
expect((await read(config, context))?.metadata?.messageCount).toBe(2)
}
)
it('reads the transcript without downloading unused card payloads', async () => {
mode = 'projected'
const doc = await read()
expect(doc?.content).toMatch(/Older decision:[\s\S]*New decision:/)
})
it.each([401, 403])('preserves HTTP %i authorization failures', async (status) => {
@@ -203,7 +391,17 @@ describe('Google Chat connector over real HTTP', () => {
it.each(['maxMessages', 'maxSpaces', 'lookbackDays'])(
'rejects invalid %s before contacting the provider',
async (field) => {
for (const value of ['Infinity', '0.5', '9007199254740992']) {
for (const value of [
'Infinity',
'0.5',
'1.0000000000000001',
'0.99999999999999999',
'1.0000000000000001e2',
'1e-400',
'9007199254740992',
true,
{},
]) {
const result = await googleChatConnector.validateConfig(TOKEN, { [field]: value })
expect(result.valid, `${field}=${value}`).toBe(false)
}
@@ -53,7 +53,7 @@ beforeEach(() => {
if (url.includes('/spaces?')) {
return jsonResponse({ spaces: listedSpaces, nextPageToken: listNextPageToken })
}
if (url.endsWith(`/${SPACE_NAME}`)) return jsonResponse(fetchedSpace)
if (new URL(url).pathname.endsWith(`/${SPACE_NAME}`)) return jsonResponse(fetchedSpace)
return jsonResponse({ error: { message: 'not found' } }, 404)
})
vi.stubGlobal('fetch', fetchMock)
@@ -94,11 +94,15 @@ describe('google-chat change detection', () => {
const runTwo: Record<string, unknown> = {}
const secondPage = await googleChatConnector.listDocuments('token', {}, undefined, runTwo)
// Stable within a run, so the hydrated document keeps the stub's hash …
const hydrated = await googleChatConnector.getDocument('token', {}, SPACE_NAME, runOne)
expect(hydrated?.contentHash).toBe(firstPage.documents[0].contentHash)
if (!hydrated) throw new Error('Expected a space document')
expect(
googleChatConnector.matchContentHash?.(
firstPage.documents[0].contentHash,
hydrated.contentHash
)
).toBe('current')
// … and different on the next run, so content never silently goes stale.
expect(secondPage.documents[0].contentHash).not.toBe(firstPage.documents[0].contentHash)
})
})
+91 -23
View File
@@ -1,6 +1,8 @@
import { createHash } from 'node:crypto'
import { createLogger } from '@sim/logger'
import { getErrorMessage } from '@sim/utils/errors'
import { generateId } from '@sim/utils/id'
import Decimal from 'decimal.js'
import {
readResponseJsonWithLimit,
readResponseToBufferWithLimit,
@@ -28,6 +30,10 @@ const MS_PER_DAY = 24 * 60 * 60 * 1000
const MAX_MESSAGE_PAGES = 200
const MAX_MESSAGE_RESPONSE_BYTES = 32 * 1024 * 1024
const MAX_SPACE_RESPONSE_BYTES = 8 * 1024 * 1024
const SPACE_FIELDS =
'name,displayName,spaceType,spaceUri,lastActiveTime,spaceDetails(description,guidelines)'
const MESSAGE_FIELDS =
'nextPageToken,messages(createTime,sender(name,displayName),text,fallbackText)'
/** `syncContext` key holding the spaces this run listed, keyed by resource name. */
const SPACE_CACHE_KEY = 'googleChatSpaces'
@@ -47,7 +53,6 @@ interface Space {
displayName?: string
spaceType?: 'SPACE_TYPE_UNSPECIFIED' | 'SPACE' | 'GROUP_CHAT' | 'DIRECT_MESSAGE'
spaceUri?: string
createTime?: string
/** Output only. Timestamp of the last message in the space. */
lastActiveTime?: string
spaceDetails?: { description?: string; guidelines?: string }
@@ -62,7 +67,6 @@ interface SpacesListResponse {
interface ChatUser {
name?: string
displayName?: string
type?: 'TYPE_UNSPECIFIED' | 'HUMAN' | 'BOT'
}
/**
@@ -72,10 +76,8 @@ interface ChatUser {
* https://developers.google.com/workspace/chat/api/reference/rest/v1/spaces.messages
*/
interface ChatMessage {
name: string
sender?: ChatUser
createTime?: string
lastUpdateTime?: string
text?: string
fallbackText?: string
}
@@ -115,11 +117,23 @@ function buildSpacesFilter(sourceConfig: Record<string, unknown>): string | unde
function resolvePositiveInteger(value: unknown, fallback: number, label: string): number {
if (value == null || (typeof value === 'string' && value.trim() === '')) return fallback
const error = `${label} must be a positive whole number`
const parsed = parseOptionalUnlimitedSafeInteger(value, error)
const parsed = parseOptionalUnlimitedSafeInteger(normalizeSavedInteger(value), error)
if (parsed === 0) throw new Error(error)
return parsed
}
/** Check the original decimal before converting legacy numeric strings to JavaScript numbers. */
function normalizeSavedInteger(value: unknown): unknown {
if (typeof value !== 'string' || value.trim() === '') return value
const parsed = Number(value)
if (!Number.isSafeInteger(parsed)) return value
try {
return new Decimal(value.trim()).isInteger() ? parsed : value
} catch {
return value
}
}
/** One cutoff per run keeps hydration consistent while inactive spaces age out on later syncs. */
function lookbackCutoff(days: number, syncContext?: Record<string, unknown>): string | undefined {
if (days === 0) return undefined
@@ -161,7 +175,7 @@ function spaceTitle(space: Space): string {
* A space without the field therefore falls back to a token that is stable for
* one sync run and different on the next, so the space re-hydrates every sync
* rather than going silently stale. `listDocuments` warns when that branch is
* taken, since it re-indexes the space's content on every sync.
* taken, since it re-reads the space's content on every sync.
*/
function activityToken(space: Space, syncContext?: Record<string, unknown>): string {
const lastActiveTime = space.lastActiveTime?.trim()
@@ -176,8 +190,7 @@ function activityToken(space: Space, syncContext?: Record<string, unknown>): str
}
/**
* Metadata-based change-detection hash. The window settings are folded in
* because changing them changes which messages the document contains.
* Metadata identifies the source version; the cutoff identifies this run's window.
*
* Known limitation: editing or deleting an existing message does not advance
* `lastActiveTime`, so such a change is only picked up once a newer message
@@ -186,11 +199,59 @@ function activityToken(space: Space, syncContext?: Record<string, unknown>): str
function buildContentHash(
space: Space,
maxMessages: number,
lookbackDays: number,
cutoff: string | undefined,
syncContext?: Record<string, unknown>
): string {
const token = activityToken(space, syncContext)
return `gchat:v2:${space.name}:${token}:${maxMessages}:${cutoff ?? 'all'}`
const version = createHash('sha256')
.update(
JSON.stringify([
space.name,
token,
maxMessages,
lookbackDays,
spaceTitle(space),
space.spaceType,
space.spaceUri,
space.spaceDetails?.description,
space.spaceDetails?.guidelines,
])
)
.digest('hex')
return `gchat:v3:${version}:${cutoff ? Date.parse(cutoff) : 'all'}`
}
const CONTENT_HASH =
/^gchat:v3:([0-9a-f]{64}):(all|-?\d+)(?::(none|unknown|-?\d+):([0-9a-f]{64}))?$/
/**
* Reuse a transcript while its source version is unchanged and no fetched message
* has aged out. Hydrated text equality advances metadata without re-embedding,
* using the same engine contract as Slack and Otter. Legacy hashes refresh once.
*/
function matchContentHash(candidate: string, stored: string): 'current' | 'equivalent' | 'stale' {
const next = CONTENT_HASH.exec(candidate)
const previous = CONTENT_HASH.exec(stored)
if (!next || !previous?.[4]) return 'stale'
if (next[4]) return next[4] === previous[4] ? 'equivalent' : 'stale'
if (next[1] !== previous[1]) return 'stale'
if (next[2] === previous[2]) return 'current'
if (next[2] === 'all' || previous[2] === 'all' || Number(next[2]) < Number(previous[2]))
return 'stale'
if (previous[3] === 'none') return 'current'
return previous[3] !== 'unknown' && Number(next[2]) < Number(previous[3]) ? 'current' : 'stale'
}
/** Unknown creation times force a reread; an exhausted empty window has no expiry. */
function oldestMessageTime(messages: ChatMessage[]): string {
let oldest = Number.POSITIVE_INFINITY
for (const message of messages) {
const created = Date.parse(message.createTime ?? '')
if (!Number.isFinite(created)) return 'unknown'
oldest = Math.min(oldest, created)
}
return oldest === Number.POSITIVE_INFINITY ? 'none' : String(oldest)
}
/** Records the spaces a listing page returned so `getDocument` can reuse them. */
@@ -216,6 +277,7 @@ function cachedSpace(name: string, syncContext?: Record<string, unknown>): Space
function spaceToStub(
space: Space,
maxMessages: number,
lookbackDays: number,
cutoff: string | undefined,
syncContext?: Record<string, unknown>
): ExternalDocument {
@@ -227,7 +289,7 @@ function spaceToStub(
estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
mimeType: 'text/plain',
sourceUrl: space.spaceUri,
contentHash: buildContentHash(space, maxMessages, cutoff, syncContext),
contentHash: buildContentHash(space, maxMessages, lookbackDays, cutoff, syncContext),
metadata: {
spaceName: spaceTitle(space),
spaceType: space.spaceType,
@@ -238,7 +300,8 @@ function spaceToStub(
/** Fetches a single space. Returns null on 404 (space deleted or left). */
async function fetchSpace(accessToken: string, name: string): Promise<Space | null> {
const response = await fetchWithRetry(`${CHAT_API_BASE}/${name}`, {
const params = new URLSearchParams({ fields: SPACE_FIELDS })
const response = await fetchWithRetry(`${CHAT_API_BASE}/${name}?${params}`, {
method: 'GET',
headers: chatHeaders(accessToken),
})
@@ -282,6 +345,7 @@ async function fetchSpaceMessages(
const params = new URLSearchParams({
pageSize: String(pageSize),
orderBy: 'createTime DESC',
fields: MESSAGE_FIELDS,
})
if (filter) params.set('filter', filter)
if (pageToken) params.set('pageToken', pageToken)
@@ -359,6 +423,7 @@ function formatSpaceContent(
export const googleChatConnector: ConnectorConfig = {
...googleChatConnectorMeta,
matchContentHash,
listDocuments: async (
accessToken: string,
@@ -367,7 +432,7 @@ export const googleChatConnector: ConnectorConfig = {
syncContext?: Record<string, unknown>
): Promise<ExternalDocumentList> => {
const maxSpaces = parseOptionalUnlimitedSafeInteger(
sourceConfig.maxSpaces,
normalizeSavedInteger(sourceConfig.maxSpaces),
'Max spaces must be a non-negative whole number'
)
const maxMessages = resolvePositiveInteger(
@@ -383,7 +448,10 @@ export const googleChatConnector: ConnectorConfig = {
maxSpaces > 0
? Math.min(SPACES_PAGE_SIZE, Math.max(1, maxSpaces - prevFetched))
: SPACES_PAGE_SIZE
const params = new URLSearchParams({ pageSize: String(pageSize) })
const params = new URLSearchParams({
pageSize: String(pageSize),
fields: `nextPageToken,spaces(${SPACE_FIELDS})`,
})
if (cursor) params.set('pageToken', cursor)
const filter = buildSpacesFilter(sourceConfig)
if (filter) params.set('filter', filter)
@@ -399,10 +467,8 @@ export const googleChatConnector: ConnectorConfig = {
})
if (!response.ok) {
const errorText = await response.text().catch(() => '')
logger.error('Failed to list Google Chat spaces', {
status: response.status,
error: errorText.slice(0, 500),
})
throw new Error(`Failed to list Google Chat spaces: ${response.status}`)
}
@@ -417,14 +483,16 @@ export const googleChatConnector: ConnectorConfig = {
const missingActivity = spaces.filter((space) => !space.lastActiveTime?.trim()).length
if (missingActivity > 0) {
logger.warn(
'Google Chat spaces listed without lastActiveTime; their content re-indexes every sync',
'Google Chat spaces listed without lastActiveTime; their content is fetched every sync',
{ spacesMissingLastActiveTime: missingActivity, spacesInPage: spaces.length }
)
}
cacheSpaces(spaces, syncContext)
const allDocuments = spaces.map((space) => spaceToStub(space, maxMessages, cutoff, syncContext))
const allDocuments = spaces.map((space) =>
spaceToStub(space, maxMessages, lookbackDays, cutoff, syncContext)
)
let documents = allDocuments
if (maxSpaces > 0) {
@@ -466,8 +534,8 @@ export const googleChatConnector: ConnectorConfig = {
/**
* Reuses the space the listing already fetched. Beyond saving a `spaces.get`
* per document, it is what keeps the hydrated `contentHash` identical to the
* stub's: re-reading the space here could observe a `lastActiveTime` that
* per document, it keeps the hydrated source version consistent with the
* stub: re-reading the space here could observe a `lastActiveTime` that
* advanced after the listing, storing a hash for messages this call never
* fetched and hiding them until the *next* message arrives.
*/
@@ -492,11 +560,12 @@ export const googleChatConnector: ConnectorConfig = {
* or `lookbackDays` was tightened past every message.
*/
const { content, messageCount } = formatSpaceContent(space, messages)
const stub = spaceToStub(space, maxMessages, cutoff, syncContext)
const stub = spaceToStub(space, maxMessages, lookbackDays, cutoff, syncContext)
return {
...stub,
content,
contentHash: `${stub.contentHash}:${oldestMessageTime(messages)}:${createHash('sha256').update(content).digest('hex')}`,
contentDeferred: false,
metadata: { ...stub.metadata, messageCount },
}
@@ -528,7 +597,7 @@ export const googleChatConnector: ConnectorConfig = {
}
try {
const params = new URLSearchParams({ pageSize: '1' })
const params = new URLSearchParams({ pageSize: '1', fields: 'spaces(name)' })
const filter = buildSpacesFilter(sourceConfig)
if (filter) params.set('filter', filter)
@@ -539,10 +608,9 @@ export const googleChatConnector: ConnectorConfig = {
)
if (!response.ok) {
const errorText = await response.text().catch(() => '')
return {
valid: false,
error: `Google Chat access failed: ${response.status}${errorText ? ` — ${errorText.slice(0, 200)}` : ''}`,
error: `Google Chat access failed: ${response.status}`,
}
}
+3 -1
View File
@@ -48,7 +48,9 @@ export const googleChatConnectorMeta: ConnectorMeta = {
* revision counter to key on instead, so a routine hash-gated sync cannot see
* an edit-only or delete-only change until a newer message lands.
*
* Routine syncs therefore stay cheap, and an explicit full resync re-hydrates
* Routine syncs reuse unchanged transcripts until a message leaves the lookback
* window. Rehydration with identical text advances metadata without re-indexing.
* An explicit full resync re-hydrates
* every space (one messages listing per space) so edits and deletions are
* picked up on demand. That cost is paid on every full resync, not once.
*/
@@ -4622,7 +4622,6 @@
"apps/sim/lib/knowledge/connectors/sync-database-retry.ts#RUN_HISTORY_LOCK_TIMEOUT_MS",
"apps/sim/lib/knowledge/connectors/sync-database-retry.ts#RUN_HISTORY_STATEMENT_TIMEOUT_MS",
"apps/sim/lib/knowledge/connectors/sync-engine.ts#ConnectorCredentialRevokedError",
"apps/sim/lib/knowledge/connectors/sync-primitives.ts#contentHashMatch",
"apps/sim/lib/knowledge/constants.ts#BooleanTagSlot",
"apps/sim/lib/knowledge/constants.ts#DateTagSlot",
"apps/sim/lib/knowledge/constants.ts#NumberTagSlot",