diff --git a/package-lock.json b/package-lock.json index 2995785d..4241557a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -14,7 +14,7 @@ "@agent-relay/sdk": "^8.1.2", "@agentworkforce/deploy": "^3.0.42", "@relayburn/sdk": "^3.2.0", - "@relayfile/sdk": "^0.7.23", + "@relayfile/sdk": "^0.8.10", "@xterm/addon-fit": "^0.10.0", "@xterm/addon-web-links": "^0.11.0", "@xterm/addon-webgl": "^0.18.0", @@ -4867,9 +4867,9 @@ } }, "node_modules/@relayfile/core": { - "version": "0.7.23", - "resolved": "https://registry.npmjs.org/@relayfile/core/-/core-0.7.23.tgz", - "integrity": "sha512-HH6mTdQ0U4J76puwrnjqiS0TgRguZzsaIIZhCSk66UECQIwGq3w2BtMm0Hh89M84vGyf829CDmqThfFSqyfXMw==", + "version": "0.8.10", + "resolved": "https://registry.npmjs.org/@relayfile/core/-/core-0.8.10.tgz", + "integrity": "sha512-fq5607zeDCw7XIB/q1xtNnFqgak9p+MQoqZ4UmreAZ/Fj8ywYv9WYdH5I3/kQX2bMCgQh1KRUjt0FaxPLPJhHg==", "license": "MIT", "engines": { "node": ">=18" @@ -4889,12 +4889,14 @@ } }, "node_modules/@relayfile/sdk": { - "version": "0.7.23", - "resolved": "https://registry.npmjs.org/@relayfile/sdk/-/sdk-0.7.23.tgz", - "integrity": "sha512-BNpM9bHO7S33dkd/TDX6woWIfKt/9U8o3geXtu5CLjmsThY7OeY4v8LqOa93Ewu/DisgIjXlyemKqcOXIiB3hA==", + "version": "0.8.10", + "resolved": "https://registry.npmjs.org/@relayfile/sdk/-/sdk-0.8.10.tgz", + "integrity": "sha512-16aFuXOv9vAdBz9uymY9QwdrhaahPGczrvw3U8NUp/9B6foyE2gb7sFPtXupKlY/M2hXYrKJ9RA7dubc5sHndw==", "license": "MIT", "dependencies": { - "@relayfile/core": "0.7.23" + "@relayfile/core": "0.8.10", + "ignore": "^7.0.5", + "tar": "^7.5.10" }, "engines": { "node": ">=18" diff --git a/package.json b/package.json index 012d2fd6..f7b9f332 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "@agent-relay/sdk": "^8.1.2", "@agentworkforce/deploy": "^3.0.42", "@relayburn/sdk": "^3.2.0", - "@relayfile/sdk": "^0.7.23", + "@relayfile/sdk": "^0.8.10", "@xterm/addon-fit": "^0.10.0", "@xterm/addon-web-links": "^0.11.0", "@xterm/addon-webgl": "^0.18.0", diff --git a/src/main/__tests__/integration-event-bridge.test.ts b/src/main/__tests__/integration-event-bridge.test.ts index cd5aaa70..89e005bd 100644 --- a/src/main/__tests__/integration-event-bridge.test.ts +++ b/src/main/__tests__/integration-event-bridge.test.ts @@ -33,6 +33,8 @@ type SubscribeCall = { options?: { coalesce?: 'none' | 'fire-once' coalesceMs?: number + pathScope?: string[] + from?: 'now' | 'legacy' onCoalesced?: () => void onQueueDepth?: (depth: number) => void } @@ -55,12 +57,15 @@ function integration(overrides: Partial & { function changeEvent( path: string, provider = path.split('/')[1] || 'github', - overrides: { digest?: string; occurredAt?: string; origin?: string; revision?: string } = {} + options: string | { digest?: string; occurredAt?: string; origin?: string; revision?: string } = {} ): ChangeEvent { - const slackTs = path.match(/\/(?:messages|replies)\/(\d{10})_(\d+)(?:\/|\.json$)/u) + const overrides = typeof options === 'string' ? { occurredAt: options } : options + const slackTs = provider === 'slack' + ? path.match(/\/(?:messages|replies)\/(\d{10})_(\d+)(?:\/|\.json$)/u) + : null const occurredAt = overrides.occurredAt ?? (slackTs?.[1] ? new Date(Number(`${slackTs[1]}.${slackTs[2] || '0'}`) * 1000).toISOString() - : '2026-06-04T00:00:00.000Z') + : new Date(Date.now() + 1000).toISOString()) return { id: `evt:${path}`, workspace: 'workspace-id', @@ -112,6 +117,16 @@ function changeEventWithFullData( } as ChangeEvent } +async function withMockedNow(isoTimestamp: string, fn: () => Promise): Promise { + const originalDateNow = Date.now + Date.now = () => Date.parse(isoTimestamp) + try { + return await fn() + } finally { + Date.now = originalDateNow + } +} + function makeHarness(agents = ['alice', 'bob'], options: { failSend?: boolean } = {}): { bridge: IntegrationEventBridge subscribeCalls: SubscribeCall[] @@ -171,6 +186,13 @@ async function waitForSent(harness: { sent: SentMessage[] }, count: number): Pro } } +async function waitForDropped(projectId: string, count: number): Promise { + const deadline = Date.now() + 1_000 + while ((getIntegrationEventTelemetrySnapshot().projects[projectId]?.eventsDropped || 0) < count && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)) + } +} + test('integration events route only to the targets for the matching integration path', async () => { const harness = makeHarness() @@ -189,6 +211,8 @@ test('integration events route only to the targets for the matching integration ]) assert.deepEqual(harness.subscribeCalls[0].globs, ['/github/repos/**', '/linear/issues/**']) + assert.deepEqual(harness.subscribeCalls[0].options?.pathScope, ['/github/repos/**', '/linear/issues/**']) + assert.equal(harness.subscribeCalls[0].options?.from, 'now') await harness.emit(changeEvent('/github/repos/acme/widgets.json', 'github')) assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) @@ -249,41 +273,146 @@ test('integration events watch selected relayfile mount paths', async () => { } }) - await harness.bridge.reconcile('project-1', [slackIntegration]) + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [slackIntegration]) + }) assert.deepEqual(harness.subscribeCalls[0].globs, [ - '/slack/channels/C123ABC/**', '/slack/channels/C123ABC__proj-cloud/**', '/slack/channels/D*/**', '/slack/dms/*/**', '/slack/users/*/messages/**' ]) + assert.deepEqual(harness.subscribeCalls[0].options?.pathScope, [ + '/slack/channels/C123ABC__proj-cloud/**', + '/slack/channels/D*/**', + '/slack/dms/*/**', + '/slack/users/*/messages/**' + ]) + assert.equal(harness.subscribeCalls[0].options?.from, 'now') assert.deepEqual(integrationSubscriptionSummaries([slackIntegration])[0].watches, [ - '.integrations/slack/channels/C123ABC/**', '.integrations/slack/channels/C123ABC__proj-cloud/**', '.integrations/slack/channels/D*/**', '.integrations/slack/dms/*/**', '.integrations/slack/users/*/messages/**' ]) - await harness.emit(changeEvent('/slack/channels/C123ABC__proj-cloud/messages/1713220123_001100/meta.json', 'slack')) + await harness.emit(changeEvent('/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json', 'slack')) await waitForSent(harness, 1) assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) - assert.match(harness.sent[0].input.text, /Path: \.integrations\/slack\/channels\/C123ABC__proj-cloud\/messages\/1713220123_001100\/meta\.json/u) - assert.match(harness.sent[0].input.text, /Relayfile path: \/slack\/channels\/C123ABC__proj-cloud\/messages\/1713220123_001100\/meta\.json/u) + assert.match(harness.sent[0].input.text, /Path: \.integrations\/slack\/channels\/C123ABC__proj-cloud\/messages\/1780668000_000000\/meta\.json/u) + assert.match(harness.sent[0].input.text, /Relayfile path: \/slack\/channels\/C123ABC__proj-cloud\/messages\/1780668000_000000\/meta\.json/u) + + harness.sent.splice(0) + await harness.emit(changeEvent('/slack/channels/C123ABC/messages/1780668060_000000/meta.json', 'slack')) + assert.deepEqual(harness.sent, []) harness.sent.splice(0) - await harness.emit(changeEvent('/slack/channels/C123ABC/messages/1713220124_001100/meta.json', 'slack')) + await harness.emit(changeEvent('/slack/channels/C999XYZ/messages/1780668120_000000/meta.json', 'slack')) + assert.deepEqual(harness.sent, []) + + await harness.emit(changeEvent('/slack/channels/D123ABC/messages/1780668180_000000/meta.json', 'slack')) await waitForSent(harness, 1) assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) +}) + +test('remote replayed events older than the subscription session are dropped by default', async () => { + const harness = makeHarness() + + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) + + await harness.emit(changeEvent( + '/slack/channels/C123ABC__proj-cloud/messages/1780315200_000000/meta.json', + 'slack', + '2026-06-01T12:00:00.000Z' + )) - harness.sent.splice(0) - await harness.emit(changeEvent('/slack/channels/C999XYZ/messages/1713220125_001100/meta.json', 'slack')) assert.deepEqual(harness.sent, []) + assert.deepEqual(harness.listAgentsCalls, []) + await waitForDropped('project-1', 1) + assert.equal(getIntegrationEventTelemetrySnapshot().projects['project-1']?.eventsDropped, 1) +}) - await harness.emit(changeEvent('/slack/channels/D123ABC/messages/1713220126_001100/meta.json', 'slack')) +test('remote events at or after the subscription session are still injected', async () => { + const harness = makeHarness() + + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) + + await harness.emit(changeEvent( + '/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json', + 'slack', + '2026-06-05T14:00:00.000Z' + )) + await waitForSent(harness, 1) + + assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) +}) + +test('remote events within replay skew before the subscription session are still injected', async () => { + const harness = makeHarness() + + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) + + await harness.emit(changeEvent( + '/slack/channels/C123ABC__proj-cloud/messages/1780667400_000000/meta.json', + 'slack', + '2026-06-05T13:50:00.000Z' + )) await waitForSent(harness, 1) + + assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) +}) + +test('historical download subscriptions can receive older remote events', async () => { + const harness = makeHarness() + + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + downloadHistoricalData: true, + scope: { notifyAgents: ['alice'] } + }) + ]) + }) + + await harness.emit(changeEvent( + '/slack/channels/C123ABC__proj-cloud/messages/1780315200_000000/meta.json', + 'slack', + '2026-06-01T12:00:00.000Z' + )) + await waitForSent(harness, 1) + assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) }) @@ -303,7 +432,6 @@ test('slack direct message event scope can be disabled', async () => { await harness.bridge.reconcile('project-1', [slackIntegration]) assert.deepEqual(harness.subscribeCalls[0].globs, [ - '/slack/channels/C123ABC/**', '/slack/channels/C123ABC__proj-cloud/**' ]) @@ -333,26 +461,32 @@ test('slack backfill and malformed nested message paths are not injected', async } })) - await harness.bridge.reconcile('project-1', [ - integration({ - provider: 'slack', - integrationId: 'slack-1', - mountPaths: ['/slack/channels/C123ABC__proj-cloud'], - scope: { notifyAgents: ['alice'] } - }) - ]) + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) await harness.emit({ ...changeEvent(stalePath, 'slack'), occurredAt: '2026-06-05T14:14:57.314Z' }) assert.deepEqual(harness.sent, []) + await waitForDropped('project-1', 1) + assert.equal(getIntegrationEventTelemetrySnapshot().projects['project-1']?.eventsDropped, 1) await harness.emit(changeEvent( '/slack/channels/C123ABC__proj-cloud/messages/1780668181_544139/slack/channels/C123ABC__proj-cloud/messages/1780668181_544139/meta.json', 'slack' )) assert.deepEqual(harness.sent, []) + await waitForDropped('project-1', 2) + assert.equal(getIntegrationEventTelemetrySnapshot().projects['project-1']?.eventsDropped, 2) } finally { await harness.bridge.close('project-1') if (originalHome === undefined) { @@ -387,14 +521,16 @@ test('slack thread reply events include local message text in the injected syste } })) - await harness.bridge.reconcile('project-1', [ - integration({ - provider: 'slack', - integrationId: 'slack-1', - mountPaths: ['/slack/channels/C123ABC__proj-cloud'], - scope: { notifyAgents: ['alice'] } - }) - ]) + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) await harness.emit(changeEvent(replyPath, 'slack')) await waitForSent(harness, 1) @@ -417,14 +553,16 @@ test('remote slack events include expanded message text before local mount sync const harness = makeHarness(['alice']) const replyPath = '/slack/channels/C123ABC__proj-cloud/threads/1780667635_192799/replies/1780668181_544139.json' - await harness.bridge.reconcile('project-1', [ - integration({ - provider: 'slack', - integrationId: 'slack-1', - mountPaths: ['/slack/channels/C123ABC__proj-cloud'], - scope: { notifyAgents: ['alice'] } - }) - ]) + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-1', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['alice'] } + }) + ]) + }) await harness.emit(changeEventWithFullData(replyPath, 'slack', { provider: 'slack', @@ -485,6 +623,37 @@ test('slack local event context rejects traversal paths', async () => { } }) +test('historical replay allowance is scoped to the matching integration', async () => { + const harness = makeHarness() + + await withMockedNow('2026-06-05T14:00:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [ + integration({ + provider: 'slack', + integrationId: 'slack-history', + mountPaths: ['/slack/channels'], + downloadHistoricalData: true, + scope: { notifyAgents: ['alice'] } + }), + integration({ + provider: 'slack', + integrationId: 'slack-live', + mountPaths: ['/slack/channels/C123ABC__proj-cloud'], + scope: { notifyAgents: ['bob'] } + }) + ]) + }) + + await harness.emit(changeEvent( + '/slack/channels/C123ABC__proj-cloud/messages/1780315200_000000/meta.json', + 'slack', + '2026-06-01T12:00:00.000Z' + )) + await waitForSent(harness, 1) + + assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice']) +}) + test('local fallback watchers are disabled when historical download is off', () => { const roots = localWatchRootsFor( 'workspace-id', @@ -699,7 +868,9 @@ test('resource alias mount paths inject the same relative event only once', asyn } }) - await harness.bridge.reconcile('project-1', [chatIntegration]) + await withMockedNow('2026-06-04T21:10:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [chatIntegration]) + }) await harness.emit(changeEvent('/chat/channels/C123ABC/threads/1780607825_485189/replies/1780611452_510669.json', 'chat')) await harness.emit(changeEvent('/chat/channels/C123ABC__proj-cloud/threads/1780607825_485189/replies/1780611452_510669.json', 'chat')) @@ -718,7 +889,9 @@ test('resource alias mount paths with the same revision inject one logical chang } }) - await harness.bridge.reconcile('project-1', [slackIntegration]) + await withMockedNow('2026-06-04T21:10:00.000Z', async () => { + await harness.bridge.reconcile('project-1', [slackIntegration]) + }) await harness.emit(changeEvent( '/slack/channels/C123ABC/messages/1780607825_485189/meta.json', diff --git a/src/main/integration-event-bridge.ts b/src/main/integration-event-bridge.ts index 0a778169..ba8693e9 100644 --- a/src/main/integration-event-bridge.ts +++ b/src/main/integration-event-bridge.ts @@ -21,6 +21,7 @@ const INTEGRATION_EVENT_SCOPES = ['relayfile:fs:read:/**'] const PROJECT_INTEGRATIONS_LINK_NAME = '.integrations' const RECENT_INJECTION_TTL_MS = 10_000 const RECENT_LOGICAL_CHANGE_TTL_MS = 10 * 60_000 +const REPLAY_SKEW_TOLERANCE_MS = 15 * 60_000 const INTEGRATION_EVENT_LOG_PATH = join(homedir(), '.agentworkforce', 'pear', 'integration-events.log') const AGGREGATED_WARNING_REPEAT_EVERY = 25 const MAX_AGGREGATED_WARNING_KEYS = 256 @@ -52,6 +53,7 @@ type SubscriptionSpec = { eventPathGlobs: string[] watches: WatchRegistration[] targets: DeliveryTargets + allowHistoricalReplay: boolean } type ProjectSubscription = { @@ -97,6 +99,8 @@ type RelayfileEventClient = { options?: { coalesce?: 'none' | 'fire-once' coalesceMs?: number + pathScope?: string[] + from?: 'now' | 'legacy' onCoalesced?: () => void onQueueDepth?: (depth: number) => void } @@ -116,6 +120,14 @@ type IntegrationEventBridgeDeps = { getWorkspaceHandle?: () => Promise } +type EventDeliverySource = 'remote' | 'local-mount' + +type EventInjectionOptions = { + source: EventDeliverySource + subscriptionStartedAtMs: number + localMountWorkspaceId: string +} + type EventWorkspaceHandleCache = { apiUrl: string accountKey: string @@ -211,6 +223,10 @@ function dedupeStrings(values: string[]): string[] { return Array.from(new Set(values.map((value) => value.trim()).filter(Boolean))).sort() } +function sameStringList(left: string[], right: string[]): boolean { + return left.length === right.length && left.every((value, index) => value === right[index]) +} + function toRelayfileProvider(provider: string): string { const normalized = provider.trim().toLowerCase() return normalized === 'gmail' ? 'google-mail' : normalized @@ -271,16 +287,7 @@ function canonicalMountPaths(integration: ConnectedIntegration): string[] { if (rootLevel) return `/${provider}${rootLevel[1] ?? ''}` return path }) - return dedupeStrings([ - ...mountPaths, - ...mountPaths.map((path) => slackChannelIdFallbackMountPath(provider, path)).filter((path): path is string => path !== null) - ]) -} - -function slackChannelIdFallbackMountPath(provider: string, path: string): string | null { - if (provider !== 'slack') return null - const match = path.match(/^\/slack\/channels\/([^/_][^/]*)__[^/]+$/u) - return match?.[1] ? `/slack/channels/${match[1]}` : null + return dedupeStrings(mountPaths) } function watchGlobForPath(path: string): string { @@ -361,7 +368,8 @@ function subscriptionSpecsFor(integrations: ConnectedIntegration[]): Subscriptio glob, coalesceMs: 750 })), - targets: deliveryTargetsFor([integration]) + targets: deliveryTargetsFor([integration]), + allowHistoricalReplay: integration.downloadHistoricalData === true } }).filter((spec) => spec.watches.length > 0) } @@ -482,6 +490,9 @@ function createWorkspaceScopedEventClient( const pendingByPath = new Map>() const coalesceMs = Math.max(0, Math.floor(options?.coalesceMs ?? 750)) const shouldCoalesce = (options?.coalesce ?? 'fire-once') !== 'none' + const pathScope = options?.pathScope?.length && !sameStringList(options.pathScope, globs) + ? options.pathScope + : null const dispatch = (event: FilesystemEvent): void => { if (!active) return @@ -505,6 +516,7 @@ function createWorkspaceScopedEventClient( if (!active || !shouldPublishFilesystemEvent(event)) return const path = event.path.startsWith('/') ? event.path : `/${event.path}` if (!globs.some((glob) => globMatchesPath(glob, path))) return + if (pathScope && !pathScope.some((glob) => globMatchesPath(glob, path))) return if (!shouldCoalesce) { dispatch({ ...event, path }) @@ -541,12 +553,16 @@ function createWorkspaceScopedEventClient( } logIntegrationEvent('remote stream starting', { workspaceId, - globs + globs, + pathScope: options?.pathScope, + from: options?.from ?? 'now' }) sync = new RelayFileSync({ client, workspaceId, token, + from: options?.from ?? 'now', + paths: options?.pathScope?.length ? options.pathScope : globs, onPollingFallback: (info) => { warnIntegrationEventAggregated( `remote stream polling fallback:${workspaceId}`, @@ -880,6 +896,23 @@ function specsForEvent(event: ChangeEvent, specs: SubscriptionSpec[]): Subscript ) } +function eventOccurredAtMs(event: ChangeEvent): number | null { + const value = Date.parse(event.occurredAt) + return Number.isFinite(value) ? value : null +} + +function historicalRemoteReplayAllowedSpecs( + event: ChangeEvent, + matchedSpecs: SubscriptionSpec[], + options: EventInjectionOptions +): SubscriptionSpec[] { + if (options.source !== 'remote') return matchedSpecs + const occurredAtMs = eventOccurredAtMs(event) + if (occurredAtMs === null) return matchedSpecs + if (occurredAtMs >= options.subscriptionStartedAtMs - REPLAY_SKEW_TOLERANCE_MS) return matchedSpecs + return matchedSpecs.filter((spec) => spec.allowHistoricalReplay) +} + function longestMatchingMountPath(path: string, spec: SubscriptionSpec): string | null { return spec.mountPaths .filter((mountPath) => pathIsInsideMount(path, mountPath)) @@ -1236,6 +1269,7 @@ export class IntegrationEventBridge { provider: spec.provider, mountPaths: spec.mountPaths, eventPathGlobs: spec.eventPathGlobs, + allowHistoricalReplay: spec.allowHistoricalReplay, targets: spec.targets })) }) @@ -1244,16 +1278,21 @@ export class IntegrationEventBridge { await this.close(projectId) const subscriptions: Subscription[] = [] try { + const remoteSubscriptionStartedAtMs = Date.now() logIntegrationEvent('subscribing', { projectId, workspaceId: handle.workspaceId, localMountWorkspaceId: handle.localMountWorkspaceId, + // temporary-pending-SDK-contract: Relayfile WS currently replays + // recent catch-up events without a from=now/path-scope contract. + remoteSubscriptionStartedAt: new Date(remoteSubscriptionStartedAtMs).toISOString(), globs: watches.map((watch) => watch.glob), specs: specs.map((spec) => ({ integrationId: spec.integrationId, provider: spec.provider, mountPaths: spec.mountPaths, eventPathGlobs: spec.eventPathGlobs, + allowHistoricalReplay: spec.allowHistoricalReplay, targets: targetLabels(spec.targets) })) }) @@ -1268,7 +1307,11 @@ export class IntegrationEventBridge { type: event.type, path: event.resource.path }) - void this.injectEvent(projectId, event, specs, handle.localMountWorkspaceId).catch((error) => { + void this.injectEvent(projectId, event, specs, { + source: 'remote', + subscriptionStartedAtMs: remoteSubscriptionStartedAtMs, + localMountWorkspaceId: handle.localMountWorkspaceId + }).catch((error) => { const errorMessage = toErrorMessage(error) warnIntegrationEventAggregated( `event delivery failed:${projectId}`, @@ -1284,6 +1327,8 @@ export class IntegrationEventBridge { { coalesce: 'fire-once', coalesceMs: Math.max(...watches.map((watch) => watch.coalesceMs), 750), + pathScope: watches.map((watch) => watch.glob), + from: 'now', onCoalesced: () => incrementIntegrationEventCounter(projectId, 'eventsCoalesced'), onQueueDepth: (depth) => setIntegrationEventGauge(projectId, 'queueDepth', depth) } @@ -1302,7 +1347,11 @@ export class IntegrationEventBridge { path: event.resource.path, source: 'local-mount' }) - void this.injectEvent(projectId, event, specs, handle.localMountWorkspaceId).catch((error) => { + void this.injectEvent(projectId, event, specs, { + source: 'local-mount', + subscriptionStartedAtMs: remoteSubscriptionStartedAtMs, + localMountWorkspaceId: handle.localMountWorkspaceId + }).catch((error) => { const errorMessage = toErrorMessage(error) warnIntegrationEventAggregated( `local event delivery failed:${projectId}`, @@ -1355,9 +1404,9 @@ export class IntegrationEventBridge { projectId: string, event: ChangeEvent, specs: SubscriptionSpec[], - localMountWorkspaceId: string + options: EventInjectionOptions ): Promise { - if (!await shouldNotifyRelayfileChange(event, localMountWorkspaceId)) { + if (!await shouldNotifyRelayfileChange(event, options.localMountWorkspaceId)) { incrementIntegrationEventCounter(projectId, 'eventsDropped') logIntegrationEvent('skipped filtered path', { projectId, @@ -1367,15 +1416,28 @@ export class IntegrationEventBridge { return } - const matchedSpecs = specsForEvent(event, specs) + const eventMatchedSpecs = specsForEvent(event, specs) + const matchedSpecs = historicalRemoteReplayAllowedSpecs(event, eventMatchedSpecs, options) if (matchedSpecs.length === 0) { incrementIntegrationEventCounter(projectId, 'eventsDropped') - logIntegrationEvent('skipped unmatched path', { - projectId, - eventId: event.id, - path: event.resource.path, - mountPaths: specs.flatMap((spec) => spec.mountPaths) - }) + if (eventMatchedSpecs.length > 0) { + logIntegrationEvent('skipped historical remote replay', { + projectId, + eventId: event.id, + path: event.resource.path, + occurredAt: event.occurredAt, + subscriptionStartedAt: new Date(options.subscriptionStartedAtMs).toISOString(), + replaySkewToleranceMs: REPLAY_SKEW_TOLERANCE_MS, + temporaryPendingSdkContract: true + }) + } else { + logIntegrationEvent('skipped unmatched path', { + projectId, + eventId: event.id, + path: event.resource.path, + mountPaths: specs.flatMap((spec) => spec.mountPaths) + }) + } return } @@ -1428,7 +1490,7 @@ export class IntegrationEventBridge { }) return } - const contextLines = await eventContextLines(event, localMountWorkspaceId) + const contextLines = await eventContextLines(event, options.localMountWorkspaceId) logIntegrationEvent('injecting', { projectId, eventId: event.id,