diff --git a/src/main/integrations.test.ts b/src/main/integrations.test.ts index a4a8ca47..fb5381bd 100644 --- a/src/main/integrations.test.ts +++ b/src/main/integrations.test.ts @@ -35,7 +35,9 @@ const mock = vi.hoisted(() => { provider: 'slack', integrationId: 'slack-integration-1', scope: {}, - mountPaths: ['/slack/channels'], + // Narrow channel mount → a writeback command root is mounted, so the default + // fixture is "active" (the integrations-update snippet now omits idle ones). + mountPaths: ['/slack/channels/C123'], connectedAt: '2026-06-05T00:00:00.000Z', notifyAgent: true, subscribeAgent: false, @@ -315,11 +317,13 @@ vi.mock('./relay-workspace', () => ({ })) import { IntegrationsManager, localSyncMountPathsForIntegration } from './integrations' +// Mocked module (see vi.mock('./integration-event-bridge') above): importing the +// stubbed function lets a test override what subscription summaries it returns. +import { integrationSubscriptionSummaries } from './integration-event-bridge' type SystemMessageSnippetBuilder = { buildSystemMessageSnippet( - integrations: Array[0]>, - subscriptionsReady: boolean + integrations: Array[0]> ): string } @@ -982,7 +986,7 @@ describe('IntegrationsManager', () => { notifyAgent: true, subscribeAgent: true, downloadHistoricalData: false - }], true) + }]) expect(message).toContain('create writeback files under .integrations/slack/channels/C123/messages') expect(message).toContain('Writeback command roots are mounted at .integrations/slack/channels/C123/messages') @@ -1002,7 +1006,7 @@ describe('IntegrationsManager', () => { notifyAgent: true, subscribeAgent: true, downloadHistoricalData: false - }], true) + }]) expect(message).toContain('create writeback files under .integrations/linear/issues') expect(message).toContain('Writeback command roots are mounted at .integrations/linear/issues') @@ -1020,12 +1024,47 @@ describe('IntegrationsManager', () => { notifyAgent: true, subscribeAgent: true, downloadHistoricalData: true - }], true) + }]) expect(message).toContain('not locally poll-mounted') expect(message).toContain('.integrations/slack/channels/C023') }) + it('treats a subscribed-but-otherwise-idle integration as active (listening)', () => { + const manager = new IntegrationsManager() as unknown as SystemMessageSnippetBuilder + // No history sync and a base mount path (no narrow writeback command roots), + // so the integration is active only by virtue of "listening" — a registered + // event subscription. + const idleSubscribedLinear = { + provider: 'linear', + integrationId: 'linear-integration-1', + scope: {}, + mountPaths: ['/linear'], + connectedAt: '2026-06-05T00:00:00.000Z', + notifyAgent: true, + subscribeAgent: true, + downloadHistoricalData: false + } + + // Without an active subscription summary the gate suppresses the message: + // not syncing, not actionable, not listening. + vi.mocked(integrationSubscriptionSummaries).mockReturnValueOnce([]) + expect(manager.buildSystemMessageSnippet([idleSubscribedLinear])).toBe('') + + // With a subscription summary present the integration is listening, so the + // message renders and lists the active subscription. + vi.mocked(integrationSubscriptionSummaries).mockReturnValue([ + { provider: 'linear', watches: ['.integrations/linear/**'], targets: [] } + ]) + const message = manager.buildSystemMessageSnippet([idleSubscribedLinear]) + expect(message).toContain('') + expect(message).toContain('linear') + expect(message).toContain('Active integration event subscriptions for this project') + expect(message).toContain('file changes at .integrations/linear/**') + + vi.mocked(integrationSubscriptionSummaries).mockReturnValue([]) + }) + it('does not block updateScope on integration state sync', async () => { let finishMountReconcile!: () => void mock.setMountReconcilePromise(new Promise((resolve) => { @@ -1171,8 +1210,9 @@ describe('IntegrationsManager', () => { // Hermetic snippet text: background cloud hydration re-adds integrations // to the store and changes the rendered text between broadcasts (the // separate staged-churn case the cooldown test covers). Block hydration - // and pin the store empty so the text is identical across reconciles. - mock.store.projects[0].integrations = [] + // and pin a single active integration so the text is identical across + // reconciles (an idle/empty project now produces no snippet at all). + mock.store.projects[0].integrations = [makeIntegration('slack', ['/slack/channels/C123'])] mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) mock.brokerManager.listAgents.mockResolvedValue([ { name: 'claude-1', projectId: 'project-1' }, @@ -1204,7 +1244,7 @@ describe('IntegrationsManager', () => { it('sends an unchanged update to a newly arrived agent without re-sending to existing agents', async () => { vi.useFakeTimers() - mock.store.projects[0].integrations = [] + mock.store.projects[0].integrations = [makeIntegration('slack', ['/slack/channels/C123'])] mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) const manager = new IntegrationsManager() @@ -1232,10 +1272,10 @@ describe('IntegrationsManager', () => { provider: 'linear', integrationId: 'linear-integration-1', scope: {}, - mountPaths: ['/linear'], + mountPaths: ['/linear/issues'], connectedAt: '2026-06-05T00:00:00.000Z', notifyAgent: true, - subscribeAgent: false, + subscribeAgent: true, downloadHistoricalData: false, visibleInProject: true }] @@ -1271,10 +1311,10 @@ describe('IntegrationsManager', () => { provider: 'linear', integrationId: 'linear-integration-1', scope: {}, - mountPaths: ['/linear'], + mountPaths: ['/linear/issues'], connectedAt: '2026-06-05T00:00:00.000Z', notifyAgent: true, - subscribeAgent: false, + subscribeAgent: true, downloadHistoricalData: false, visibleInProject: true }] @@ -1286,10 +1326,11 @@ describe('IntegrationsManager', () => { manager.recordSpawnInstructionDelivery('project-1', 'codex-1') // Simulate staged churn: the store changes right after the spawn, so the - // broadcast text differs from the embedded snippet. + // broadcast text differs from the embedded snippet (historical download + // flips on, which rewrites the history clause). mock.store.projects[0].integrations = [{ ...mock.store.projects[0].integrations[0], - mountPaths: ['/linear/issues'] + downloadHistoricalData: true }] await manager.notifyAgentState('project-1') await vi.advanceTimersByTimeAsync(5_000) @@ -1309,7 +1350,7 @@ describe('IntegrationsManager', () => { it('holds a changed update inside the rebroadcast cooldown and sends the latest text once', async () => { vi.useFakeTimers() - mock.store.projects[0].integrations = [] + mock.store.projects[0].integrations = [makeIntegration('slack', ['/slack/channels/C123'])] mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) const manager = new IntegrationsManager() @@ -1330,10 +1371,10 @@ describe('IntegrationsManager', () => { provider: 'linear', integrationId: 'linear-integration-1', scope: {}, - mountPaths: ['/linear'], + mountPaths: ['/linear/issues'], connectedAt: '2026-06-05T00:00:00.000Z', notifyAgent: true, - subscribeAgent: false, + subscribeAgent: true, downloadHistoricalData: false, visibleInProject: true }] @@ -1346,12 +1387,41 @@ describe('IntegrationsManager', () => { expect(updateTexts()[1]).toContain('linear') }) - it('does not re-send an unchanged integrations update after the dedupe window lapses', async () => { + it('broadcasts one clearing update when the last active integration is removed', async () => { vi.useFakeTimers() - // Pin the state so the rendered text is identical across reconciles - // (background hydration would otherwise change it, which is the - // legitimate-rebroadcast case). + mock.store.projects[0].integrations = [makeIntegration('slack', ['/slack/channels/C123'])] + mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) + mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) + const manager = new IntegrationsManager() + const updateTexts = (): string[] => integrationUpdateSendInputs().map((input) => input.text) + + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + expect(updateTexts()).toHaveLength(1) + expect(updateTexts()[0]).toContain('.integrations/slack/channels/C123') + + // Last integration disconnected: agents already told about it must get one + // clearing update rather than silence (otherwise they keep stale context). mock.store.projects[0].integrations = [] + await vi.advanceTimersByTimeAsync(60_000) // clear the rebroadcast cooldown + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(1_000) + expect(updateTexts()).toHaveLength(2) + expect(updateTexts()[1]).toContain('- none') + + // A further idle reconcile stays silent — the prior context was cleared once. + await vi.advanceTimersByTimeAsync(60_000) + await manager.notifyAgentState('project-1') + await vi.advanceTimersByTimeAsync(60_000) + expect(updateTexts()).toHaveLength(2) + }) + + it('does not re-send an unchanged integrations update after the dedupe window lapses', async () => { + vi.useFakeTimers() + // Pin a single active integration so the rendered text is identical across + // reconciles (background hydration would otherwise change it, which is the + // legitimate-rebroadcast case; an idle/empty project now yields no snippet). + mock.store.projects[0].integrations = [makeIntegration('slack', ['/slack/channels/C123'])] mock.getAccountWorkspaceId.mockRejectedValue(new Error('offline')) mock.brokerManager.listAgents.mockResolvedValue([{ name: 'claude-1', projectId: 'project-1' }] as never) const manager = new IntegrationsManager() diff --git a/src/main/integrations.ts b/src/main/integrations.ts index 327d4d99..eb1ad8a5 100644 --- a/src/main/integrations.ts +++ b/src/main/integrations.ts @@ -14,8 +14,11 @@ import { integrationMountRootForWorkspace, MAX_LOCAL_INTEGRATION_MOUNT_PATHS } from './integration-mounts' -import { parseWritableResources as parsePackageWritableResources } from '@agent-relay/integration-prompts' -import type { WritableResourceDescriptor } from '@agent-relay/integration-prompts' +import { + fullInjectInstructions as buildIntegrationsUpdateSnippet, + parseWritableResources as parsePackageWritableResources +} from '@agent-relay/integration-prompts' +import type { IntegrationDescriptor, WritableResourceDescriptor } from '@agent-relay/integration-prompts' import { canListRemoteDirectoryForMountPaths, canShowRemoteDirectoryEntryForMountPaths, @@ -894,10 +897,6 @@ export class IntegrationsManager { // newly spawned agent receives the current update without re-sending it to // everyone else. private lastAgentSystemMessages = new Map>() - // Subscription readiness from the last reconcile, used so spawn-embedded - // snippets render with the same readiness clause the next broadcast will - // use (a hardcoded `true` made the texts differ and re-notified the agent). - private lastSubscriptionsReady = new Map() // When each project's last broadcast went out. Mount-path discovery and // subscription readiness converge in stages (e.g. bare → suffixed Slack // channel ids), each stage changing the rendered text slightly; without a @@ -906,6 +905,12 @@ export class IntegrationsManager { // Snippet most recently embedded into spawn instructions, per project, so // the spawn path can record it as already-delivered for the new agent. private lastSpawnInstructionSnippets = new Map() + // Projects that have broadcast (or spawned with) a non-empty integrations + // update. When such a project transitions to having no active integrations + // (e.g. the last one is disconnected), we broadcast one clearing update so + // already-notified agents stop relying on the stale context — idle context is + // only suppressed for spawn/no-op, not for state changes that clear it. + private projectsWithActiveSnippet = new Set() private catalogCache: IntegrationAdapter[] | null = null private catalogFetchedAt = 0 private localMountCloudHydrationStartedAt = 0 @@ -2238,15 +2243,23 @@ export class IntegrationsManager { systemMessageOptions: IntegrationSystemMessageOptions = {} ): Promise { const integrations = this.visibleIntegrationsForProject(projectId) - const subscriptionsReady = await this.syncEventSubscriptions(projectId) - this.lastSubscriptionsReady.set(projectId, subscriptionsReady) + await this.syncEventSubscriptions(projectId) if (notifyAgent) { - this.scheduleSystemMessage( - projectId, - this.buildSystemMessageSnippet(integrations, subscriptionsReady), - systemMessageOptions - ) + const snippet = this.buildSystemMessageSnippet(integrations) + if (snippet) { + this.projectsWithActiveSnippet.add(projectId) + this.scheduleSystemMessage(projectId, snippet, systemMessageOptions) + } else if (this.projectsWithActiveSnippet.delete(projectId)) { + // Active → none transition: tell already-notified agents the prior + // integration context is gone, rather than leaving them to attempt + // stale writebacks or wait on subscriptions that no longer exist. + this.scheduleSystemMessage( + projectId, + buildIntegrationsUpdateSnippet([]), + systemMessageOptions + ) + } } void this.syncLocalIntegrationState(projectId).catch((error) => { @@ -2259,96 +2272,92 @@ export class IntegrationsManager { await this.safeUpdateMountPaths(projectId, this.mountPathsFor(projectId)) } - private buildSystemMessageSnippet(integrations: ConnectedIntegration[], subscriptionsReady: boolean): string { - const lines = [ - '', - 'The user has connected the following integrations to this project:' - ] - - if (integrations.length === 0) { - lines.push('- none') - } else { - for (const integration of integrations) { - const scopeLabels = collectScopeLabels(integration.scope) - const scopeSummary = scopeLabels.length > 0 ? scopeLabels.join(', ') : 'all configured scope' - const mountPaths = this.canonicalMountPathsForIntegration(integration) - .map(projectIntegrationPathForRelayfilePath) - const discoveryPath = projectIntegrationPathForRelayfilePath(discoveryMountPathForProvider(integration.provider)) - const writebackCommandPaths = writebackCommandMountPathsForIntegration(integration) - .map(projectIntegrationPathForRelayfilePath) - const liveContextPaths = integration.downloadHistoricalData === true - ? [] - : slackThreadContextMountPathsForIntegration(integration) - .map(projectIntegrationPathForRelayfilePath) - const historyEnabled = integration.downloadHistoricalData === true - const writebackPaths = historyEnabled - ? mountPaths.join(', ') || 'the configured provider paths' - : writebackCommandPaths.join(', ') - const writebackInstruction = writebackPaths - ? `create writeback files under ${writebackPaths}, not under discovery` - : 'select narrower resources before creating local writeback files; discovery is schema-only' - const skippedHistoryPaths = skippedHistoricalMountPathsForIntegration(integration) - .map(projectIntegrationPathForRelayfilePath) - const skippedCappedPaths = skippedCappedLocalSyncMountPathsForIntegration(integration) + // Builds the narrative via the shared + // @agent-relay/integration-prompts builder (fullInjectInstructions) instead of + // a hand-maintained copy. Returns '' when no integration is currently active + // (see buildActiveIntegrationDescriptors) so callers can skip the message + // entirely — a visible-but-idle integration adds only noise at spawn. + private buildSystemMessageSnippet(integrations: ConnectedIntegration[]): string { + const descriptors = this.buildActiveIntegrationDescriptors(integrations) + if (descriptors.length === 0) return '' + return buildIntegrationsUpdateSnippet(descriptors) + } + + // Maps connected integrations to the package's IntegrationDescriptor shape, + // keeping only the ones the agent actually needs to know about at spawn: + // - syncing: historical download is enabled, OR + // - listening: a registered event subscription exists, OR + // - actionable: narrow writeback command roots are mounted. + // Visible integrations with none of the above (no sync, no events, no narrow + // resources) are omitted; if nothing qualifies the caller suppresses the + // whole message. + private buildActiveIntegrationDescriptors( + integrations: ConnectedIntegration[] + ): IntegrationDescriptor[] { + const descriptors: IntegrationDescriptor[] = [] + for (const integration of integrations) { + const historyEnabled = integration.downloadHistoricalData === true + const eventScopePaths = this.canonicalMountPathsForIntegration(integration) + .map(projectIntegrationPathForRelayfilePath) + const writebackCommandPaths = writebackCommandMountPathsForIntegration(integration) + .map(projectIntegrationPathForRelayfilePath) + // Listening is derived per integration from its own config (subscribeAgent + // + event globs), independent of whether Relayfile has finished + // registering it. Scoping the summary to this single integration avoids + // marking every integration of a provider active when only one of several + // (e.g. multiple Slack workspaces) actually subscribes. + const subscriptions = integrationSubscriptionSummaries([integration]) + + const isActive = historyEnabled + || writebackCommandPaths.length > 0 + || subscriptions.length > 0 + if (!isActive) continue + + const writebackPaths = historyEnabled ? eventScopePaths : writebackCommandPaths + const liveContextPaths = historyEnabled + ? [] + : slackThreadContextMountPathsForIntegration(integration) .map(projectIntegrationPathForRelayfilePath) - const skippedLocalPaths = dedupeStrings([...skippedHistoryPaths, ...skippedCappedPaths]) - const scopeClause = mountPaths.length > 0 - ? ` (event scope ${mountPaths.join(', ')})` - : ' (no event scope configured)' - const historyClause = historyEnabled - ? skippedLocalPaths.length > 0 - ? ` Historical download is enabled, but these provider paths are not locally poll-mounted: ${skippedLocalPaths.join(', ')}. Select fewer or narrower resources to download local history.` - : ` Historical provider records are available at ${writebackPaths}.` - : writebackPaths - ? ` Local historical provider records are not broadly downloaded. Writeback command roots are mounted at ${writebackPaths}${liveContextPaths.length > 0 ? `, and live thread context roots are mounted at ${liveContextPaths.join(', ')}` : ''}; provider context should be read on demand or through incoming events.` - : ` Local historical provider records are not downloaded. No narrow writeback command roots are mounted; select narrower resources to enable local writeback transport.` - lines.push( - `- ${integration.provider}: ${scopeSummary}${scopeClause}. Writeback schemas and examples are available at ${discoveryPath}; ${writebackInstruction}. ${historyClause}` - ) - } - } + const skippedLocalPaths = dedupeStrings([ + ...skippedHistoricalMountPathsForIntegration(integration).map(projectIntegrationPathForRelayfilePath), + ...skippedCappedLocalSyncMountPathsForIntegration(integration).map(projectIntegrationPathForRelayfilePath) + ]) - const subscriptions = integrationSubscriptionSummaries(integrations) - lines.push('') - if (!subscriptionsReady && subscriptions.length > 0) { - lines.push('Integration event subscriptions are requested for this project, but Pear could not register them with Relayfile yet.') - lines.push('Do not assume notifications will arrive until a later integrations update confirms active subscriptions. If historical provider records are mounted for an integration, read them when the user asks for current state; otherwise rely on incoming event notifications, inline event context, and the event file path when present.') - } else if (subscriptions.length === 0) { - lines.push('No integration event subscriptions are active for this project.') - } else { - lines.push('Active integration event subscriptions for this project:') - for (const subscription of subscriptions) { - const parts = [ - subscription.watches.length > 0 ? `file changes at ${subscription.watches.join(', ')}` : '', - subscription.targets.length > 0 ? `delivered to ${subscription.targets.join(', ')}` : 'delivered to all project agents' - ].filter(Boolean) - lines.push(`- ${subscription.provider}: ${parts.join('; ')}`) - } - lines.push( - 'You will receive system messages for these subscribed changes. Do not poll these integrations for background changes; wait for the event notification, then read the mounted files for context if needed.' - ) + descriptors.push({ + provider: integration.provider, + mountRoot: `${PROJECT_INTEGRATIONS_LINK_NAME}/${integration.provider}`, + writableResources: [], + discoveryRoot: projectIntegrationPathForRelayfilePath( + discoveryMountPathForProvider(integration.provider) + ), + scopeLabels: collectScopeLabels(integration.scope), + eventScopePaths, + writebackPaths, + downloadHistoricalData: historyEnabled, + liveContextPaths, + skippedLocalPaths, + subscriptions + }) } - - lines.push( - `Writeback discovery schemas and examples are mounted through ${PROJECT_INTEGRATIONS_LINK_NAME}/discovery// for connected integrations. Use those schemas to create files under the provider paths such as ${PROJECT_INTEGRATIONS_LINK_NAME}/slack/channels//messages; do not create provider records inside discovery. Historical provider records are only intentionally downloaded when historical download is enabled. Incoming webhook events do not require downloading history.`, - '' - ) - return lines.join('\n') + return descriptors } initialSpawnInstructions(projectId: string): string | undefined { const integrations = this.visibleIntegrationsForProject(projectId) if (integrations.length === 0) return undefined - const snippet = this.buildSystemMessageSnippet( - integrations, - this.lastSubscriptionsReady.get(projectId) ?? true - ) + const snippet = this.buildSystemMessageSnippet(integrations) + // No integration is currently syncing/listening/actionable — skip the spawn + // message entirely rather than inject an empty "- none" narrative. + if (!snippet) return undefined // Remember what this spawn embedded so recordSpawnInstructionDelivery // can mark the new agent as already-notified — without it, the next // project broadcast re-delivered the same setup context to the agent // that just received it in its spawn task ("Setup context received" // acknowledged twice). this.lastSpawnInstructionSnippets.set(projectId, snippet) + // A later transition to no active integrations should still clear this for + // the spawned agent (see projectsWithActiveSnippet). + this.projectsWithActiveSnippet.add(projectId) return [ 'Initial project integration context. Treat this as setup context, not as the user task.', snippet