From 80a1596a3d139274dcc53fc47462911bc21c0dbe Mon Sep 17 00:00:00 2001 From: Bernhard Wittmann Date: Mon, 28 Sep 2026 14:41:26 +0000 Subject: [PATCH] feat(core): Render Microsoft Teams agent replies progressively (no-changelog) (#39514) --- .../agent-chat-stream-consumer.test.ts | 124 +++++++++++++ .../channel-integration-contract.test.ts | 4 +- .../__tests__/helpers/replay-test-helpers.ts | 14 +- .../helpers/teams/replay-test-context.ts | 48 ++++- .../__tests__/teams-integration.test.ts | 57 +++++- .../agents/integrations/agent-chat-bridge.ts | 1 + .../integrations/agent-chat-integration.ts | 7 + .../agent-chat-stream-consumer.ts | 82 +++++++-- .../integrations/chat-integration.service.ts | 21 +++ .../teams/synthetic-integration.test.ts | 167 +++++++++++++++++- .../platforms/teams/teams-integration.ts | 66 ++++++- 11 files changed, 553 insertions(+), 38 deletions(-) diff --git a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts index d5296a6b1a0..5a2d2aff98e 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts @@ -120,6 +120,130 @@ describe('AgentChatStreamConsumer — rate-limit fallback', () => { }); }); +function isAsyncIterable(value: unknown): boolean { + return typeof (value as Record)?.[Symbol.asyncIterator] === 'function'; +} + +/** Separates streamed posts (an async iterable) from discrete ones. */ +function makeStreamingThread({ + stall = false, + settleAfterMs, + rejectAfterMs, +}: { stall?: boolean; settleAfterMs?: number; rejectAfterMs?: number } = {}) { + const streamed: unknown[] = []; + const discrete: unknown[] = []; + const post = vi.fn(async (message: unknown) => { + if (!isAsyncIterable(message)) { + discrete.push(message); + return undefined; + } + streamed.push(message); + if (settleAfterMs !== undefined) { + return await new Promise((resolve) => setTimeout(resolve, settleAfterMs)); + } + if (rejectAfterMs !== undefined) { + return await new Promise((_resolve, reject) => + setTimeout(() => reject(new Error('late failure')), rejectAfterMs), + ); + } + if (stall) return await new Promise(() => {}); + for await (const _chunk of message as AsyncIterable) { + // Drain the iterable the way a real adapter does. + } + return undefined; + }); + const thread = mock>(); + thread.post = post as unknown as typeof thread.post; + return { thread, streamed, discrete }; +} + +function makeStreamingConsumer( + options: Partial[0]> = {}, +) { + return new AgentChatStreamConsumer({ + disableStreaming: false, + logger: mock(), + postErrorToThread: vi.fn().mockResolvedValue(undefined), + handleSuspension: vi.fn().mockResolvedValue('posted'), + handleMessage: vi.fn().mockResolvedValue(true), + isIntegrationActionTool: () => true, + ...options, + }); +} + +describe('AgentChatStreamConsumer — singleStreamedRunPerTurn', () => { + const textThenMessageThenText = (): StreamChunk[] => [ + { type: 'text-delta', id: 't-1', delta: 'Before' }, + { type: 'message', message: { text: 'a card' } } as unknown as StreamChunk, + { type: 'text-delta', id: 't-2', delta: 'After' }, + ]; + + it('streams once and posts the trailing text as its own message', async () => { + const { thread, streamed, discrete } = makeStreamingThread(); + const consumer = makeStreamingConsumer({ singleStreamedRunPerTurn: true }); + + await consumer.consume(makeStream(textThenMessageThenText()), thread); + + expect(streamed).toHaveLength(1); + expect(discrete).toEqual([{ markdown: 'After' }]); + }); + + it('opens a second streamed post when the platform allows it', async () => { + const { thread, streamed, discrete } = makeStreamingThread(); + const consumer = makeStreamingConsumer(); + + await consumer.consume(makeStream(textThenMessageThenText()), thread); + + expect(streamed).toHaveLength(2); + expect(discrete).toEqual([]); + }); +}); + +describe('AgentChatStreamConsumer — silent outcome after a discrete post', () => { + it('drops trailing text that a do_not_respond outcome silenced', async () => { + const { thread, streamed, discrete } = makeStreamingThread(); + const consumer = makeStreamingConsumer({ singleStreamedRunPerTurn: true }); + + await consumer.consume( + makeStream([ + { type: 'text-delta', id: 't-1', delta: 'Before' }, + { type: 'message', message: { text: 'a card' } } as unknown as StreamChunk, + { type: 'text-delta', id: 't-2', delta: 'After' }, + { + type: 'tool-result', + toolCallId: 'tc-1', + toolName: 'teams_action', + output: { silent: true }, + }, + ]), + thread, + ); + + expect(streamed).toHaveLength(1); + expect(discrete).toEqual([]); + }); +}); + +describe('AgentChatStreamConsumer — delivery that must not stream', () => { + it('buffers a proactive send even on a streaming platform', async () => { + const { thread, streamed, discrete } = makeStreamingThread(); + const consumer = makeStreamingConsumer(); + + await consumer.consume( + makeStream([ + { type: 'text-delta', id: 't-1', delta: 'Scheduled ' }, + { type: 'text-delta', id: 't-1', delta: 'reminder' }, + ]), + thread, + // What deliverWakeResponse passes, so a failure can be retried. + { throwOnDeliveryError: true }, + ); + + expect(streamed).toEqual([]); + expect(discrete).toEqual([{ markdown: 'Scheduled reminder' }]); + }); +}); + describe('AgentChatStreamConsumer — suspension routing', () => { function makeSuspensionConsumer(disableStreaming: boolean) { const handleSuspension = vi.fn().mockResolvedValue('posted'); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/channel-integration-contract.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/channel-integration-contract.test.ts index a4290b40a59..c0dff4787ff 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/channel-integration-contract.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/channel-integration-contract.test.ts @@ -148,8 +148,10 @@ runSharedChannelIntegrationContract({ }, resourceId: TEAMS_USER_ID, firstPost: { + // A direct message reply is rendered by editing this post, so the first + // post is the placeholder and the text arrives in a later edit. type: 'message', - text: 'Got it', + text: '…', conversation: { id: TEAMS_DM_CONVERSATION_ID }, }, respondPost: { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts index 79ba2e2de9a..770054bd071 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts @@ -82,10 +82,14 @@ export class MemoryMessageContextStore implements IntegrationMessageContextStore } } -export function toStream(chunks: StreamChunk[]): AsyncGenerator { +export function toStream(chunks: StreamChunk[], gapMs = 0): AsyncGenerator { return (async function* stream() { await Promise.resolve(); - for (const chunk of chunks) yield chunk; + for (const chunk of chunks) { + // A gap lets a platform that renders on a timer actually tick. + if (gapMs) await new Promise((resolve) => setTimeout(resolve, gapMs)); + yield chunk; + } })(); } @@ -208,6 +212,8 @@ export function createReplayContextSetup(params: { integration: AgentIntegrationConfig; componentMapper?: ComponentMapper; stream?: StreamChunk[]; + /** Delay between chunks, so a timer-driven renderer ticks. */ + streamGapMs?: number; }): ReplayContextSetup { const registry = new ChatIntegrationRegistry(); registry.register(params.integrationImpl); @@ -226,11 +232,11 @@ export function createReplayContextSetup(params: { executeForChatPublished: vi.fn((config) => { selectedContext = config.messageContext ?? undefined; selectedThreadId = config.memory.threadId.id; - return toStream(stream); + return toStream(stream, params.streamGapMs); }), resumeForChat: vi.fn((config) => { selectedContext = config.messageContext ?? undefined; - return toStream(stream); + return toStream(stream, params.streamGapMs); }), // Mirrors how production wires the gate. It admits every run by default, // so a test that wants the stale branch resolves it to false. diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/replay-test-context.ts index 91b17c13dba..d0724d52e9d 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/replay-test-context.ts @@ -1,4 +1,5 @@ import type { StreamChunk } from '@n8n/agents'; +import { isRecord } from '@n8n/utils/is-record'; import type { Logger as BackendLogger } from '@n8n/backend-common'; import { generateKeyPairSync, randomUUID } from 'crypto'; import jwt from 'jsonwebtoken'; @@ -42,6 +43,10 @@ export interface TeamsReplayContext extends Omit { latestThreadId: () => string | undefined; lastPost: () => ReplayApiCall | undefined; lastEdit: () => ReplayApiCall | undefined; + /** Every activity the adapter attempted, in order, refused ones included. */ + activities: () => ReplayApiCall[]; + /** Every edit of an already-posted message, in order. */ + edits: () => ReplayApiCall[]; lastDelete: () => ReplayApiCall | undefined; lastPostedMessageId: () => string | undefined; } @@ -90,7 +95,16 @@ function buildBotFrameworkSigner() { }; } -function installTeamsApiStub(jwks: object, accessToken: string) { +/** The `streaminfo` entity that ties an activity to an open Teams stream. */ +export function streamInfo(body: Record): Record | undefined { + const entities = body.entities; + if (!Array.isArray(entities)) return undefined; + return entities.find((entity) => isRecord(entity) && entity.type === 'streaminfo') as + | Record + | undefined; +} + +function installTeamsApiStub(jwks: object, accessToken: string, succeedingEdits?: number) { const apiCalls: ReplayApiCall[] = []; const serviceUrl = new URL(TEAMS_SERVICE_URL); @@ -108,15 +122,17 @@ function installTeamsApiStub(jwks: object, accessToken: string) { // Recorded as `sendActivity` so `lastPost()` reads like the other platforms'. let nextMessageId = 1000; + let editCount = 0; const postedMessageIds: string[] = []; nock(serviceUrl.origin) .persist() .post(/\/v3\/conversations\/.+\/activities.*/) .reply(function (_uri, body) { - apiCalls.push({ - method: 'sendActivity', - body: (typeof body === 'object' && body !== null ? body : {}) as Record, - }); + const activity = (typeof body === 'object' && body !== null ? body : {}) as Record< + string, + unknown + >; + apiCalls.push({ method: 'sendActivity', body: activity }); const id = `message-${nextMessageId++}`; postedMessageIds.push(id); return [200, { id }]; @@ -138,6 +154,9 @@ function installTeamsApiStub(jwks: object, accessToken: string) { method: 'updateActivity', body: (typeof body === 'object' && body !== null ? body : {}) as Record, }); + if (succeedingEdits !== undefined && editCount++ >= succeedingEdits) { + return [429, { error: { message: 'Too many requests' } }]; + } return [200, { id: 'message-edited' }]; }); @@ -145,10 +164,15 @@ function installTeamsApiStub(jwks: object, accessToken: string) { } export async function createTeamsReplayContext( - options: { stream?: StreamChunk[] } = {}, + options: { + stream?: StreamChunk[]; + streamGapMs?: number; + /** Reject every edit past this many, so a rejection can be provoked. */ + succeedingEdits?: number; + } = {}, ): Promise { const signer = createBotFrameworkSigner(); - const stub = installTeamsApiStub(signer.jwks, signer.accessToken()); + const stub = installTeamsApiStub(signer.jwks, signer.accessToken(), options.succeedingEdits); // Dynamic imports — the chat packages are ESM-only. Production routes through // esm-loader to dodge the CJS transform; vitest loads ESM natively. @@ -166,11 +190,17 @@ export async function createTeamsReplayContext( userName: 'n8n-agent-agent-1', adapters: { teams: adapter } as unknown as Record, state: createMemoryState(), + // Mirrors what ChatIntegrationService gives a Teams connection; without it + // the adapter buffers instead of handing back to the post-and-edit path. + fallbackStreamingPlaceholderText: '…', + streamingUpdateIntervalMs: 20, }); + const integrationImpl = new TeamsIntegration(mock(), mock()); const setup = createReplayContextSetup({ chat: chat as never, - integrationImpl: new TeamsIntegration(mock(), mock()), + integrationImpl, + streamGapMs: options.streamGapMs, integration: { type: 'teams', credentialId: 'cred-teams', settings: undefined }, componentMapper: new ComponentMapper(), stream: options.stream, @@ -205,6 +235,8 @@ export async function createTeamsReplayContext( latestThreadId: setup.latestThreadId, lastPost: () => lastCall('sendActivity'), lastEdit: () => lastCall('updateActivity'), + activities: () => stub.apiCalls.filter((call) => call.method === 'sendActivity'), + edits: () => stub.apiCalls.filter((call) => call.method === 'updateActivity'), lastDelete: () => lastCall('deleteActivity'), lastPostedMessageId: () => stub.postedMessageIds.at(-1), shutdown: async () => { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/teams-integration.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/teams-integration.test.ts index 52d96cf4328..bd2552770b1 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/teams-integration.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/teams-integration.test.ts @@ -6,7 +6,10 @@ import { ConflictError } from '@/errors/response-errors/conflict.error'; import type { Agent } from '../../entities/agent.entity'; import type { AgentRepository } from '../../repositories/agent.repository'; -import type { AgentChatIntegrationContext } from '../agent-chat-integration'; +import type { + AgentChatIntegrationContext, + BridgeMessageContextParams, +} from '../agent-chat-integration'; import { TEAMS_APP_ID as CLIENT_ID, TEAMS_CLIENT_SECRET as CLIENT_SECRET, @@ -63,11 +66,14 @@ describe('TeamsIntegration', () => { }); describe('platform capabilities', () => { - it('reuses the Entra credential, stays hidden, buffers, and targets its cards', () => { + it('reuses the Entra credential, stays hidden, streams per thread, and targets its cards', () => { expect(integration).toMatchObject({ credentialTypes: ['microsoftEntraServicePrincipalApi'], internal: true, - disableStreaming: true, + // Streaming is decided for each conversation instead, because Teams + // only supports it in 1:1 chats. + disableStreaming: false, + singleStreamedRunPerTurn: true, // Left at the base-class default: full Adaptive Card payloads fit, so // no callback store is needed. needsShortCallbackData: false, @@ -237,4 +243,49 @@ describe('TeamsIntegration', () => { expect(integration.normalizeComponents(components)).toEqual(components); }); }); + + describe('streaming decisions', () => { + function contextParams(isDM: boolean, credentialId = CREDENTIAL_ID) { + const startTyping = vi.fn().mockResolvedValue(undefined); + const params = { + chat: {}, + thread: { id: 'thread-1', isDM, startTyping }, + message: {}, + integration: { type: 'teams', credentialId, settings: undefined }, + logger: mock(), + agentId: AGENT_ID, + } as unknown as BridgeMessageContextParams; + return { params, startTyping }; + } + + it('streams a direct message and shows the typing indicator', async () => { + const { params, startTyping } = contextParams(true); + + const context = await integration.createBridgeExecutionContext(params); + + expect(context.forceBuffered).toBe(false); + expect(startTyping).toHaveBeenCalledTimes(1); + await context.statusHandle?.clearBeforeResponse(); + }); + + it('buffers a group chat or channel, and still shows the typing indicator', async () => { + const { params, startTyping } = contextParams(false); + + const context = await integration.createBridgeExecutionContext(params); + + expect(context.forceBuffered).toBe(true); + expect(startTyping).toHaveBeenCalledTimes(1); + await context.statusHandle?.clearBeforeResponse(); + }); + + it('buffers a reply to a card action, because an invoke activity has no streamer', async () => { + const { params, startTyping } = contextParams(true); + + const context = await integration.createResumeExecutionContext(params); + + expect(context.forceBuffered).toBe(true); + expect(startTyping).toHaveBeenCalledTimes(1); + await context.statusHandle?.clearBeforeResponse(); + }); + }); }); diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts index 71dd03be730..61b0f7caeae 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts @@ -246,6 +246,7 @@ export class AgentChatBridge { const actionToolNamePattern = new RegExp(`^${integration.type}(_\\d+)?_action$`); this.streamConsumer = new AgentChatStreamConsumer({ disableStreaming, + singleStreamedRunPerTurn: this.integrationImpl?.singleStreamedRunPerTurn, logger: this.logger, postErrorToThread: this.postErrorToThread.bind(this), handleSuspension: this.handleSuspension.bind(this), diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts b/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts index f5af4eaf2cc..94cd6258873 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts @@ -289,6 +289,13 @@ export abstract class AgentChatIntegration { */ readonly disableStreaming: boolean = false; + /** + * True when the platform renders only one streamed run per inbound turn, so + * text following a card is posted buffered rather than streamed into the + * message that preceded the card. + */ + readonly singleStreamedRunPerTurn: boolean = false; + /** * True when this integration is an internal channel (e.g. the in-app n8n * chat) that must not appear in the public integrations catalog or the diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts b/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts index 2b42ac4f2c1..fae0351eb40 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts @@ -38,6 +38,8 @@ interface AgentChatStreamConsumerOptions { * tools returning a `silent` field must not mute the reply. */ isIntegrationActionTool?: (toolName: string) => boolean; + /** Text after the first streamed run is buffered and posted on its own. */ + singleStreamedRunPerTurn?: boolean; } interface ConsumeStreamOptions { @@ -109,6 +111,14 @@ export class AgentChatStreamConsumer { end: null, }; let streamingPost: Promise | null = null; + /** Text the platform is not yet known to have rendered. */ + let pendingText = ''; + let streamingStopped = false; + /** Only a platform that can stop streaming mid-turn re-reads the text. */ + const retainText = this.options.singleStreamedRunPerTurn === true; + /** Set when the post failed, so the retained text is posted instead. */ + let streamingPostRejected = false; + let streamingPostError: unknown; const createTextIterable = (): AsyncIterable => { const queue: string[] = []; @@ -155,8 +165,11 @@ export class AgentChatStreamConsumer { const startStreamingPost = () => { const iterable = createTextIterable(); - streamingPost = thread.post(iterable).catch(async (postError: unknown) => { - await this.options.postErrorToThread(thread, postError); + streamingPostRejected = false; + streamingPostError = undefined; + streamingPost = thread.post(iterable).catch((postError: unknown) => { + streamingPostRejected = true; + streamingPostError = postError; this.options.logger.error('[AgentChatBridge] Streaming post failed', { error: postError instanceof Error ? postError.message : String(postError), }); @@ -170,13 +183,28 @@ export class AgentChatStreamConsumer { textStream.yield = null; } if (streamingPost) { - await streamingPost; + const post = streamingPost; streamingPost = null; + if (this.options.singleStreamedRunPerTurn) streamingStopped = true; + await post; + // A post that rejected left the reply unsent — on a post-and-edit + // platform the user is looking at a placeholder. Post the retained + // text below instead of dropping it, and fall back to telling the + // user only when there is no text to deliver. + if (!streamingPostRejected) { + pendingText = ''; + } else if (!pendingText.trim()) { + await this.options.postErrorToThread(thread, streamingPostError); + } } + const text = pendingText; + pendingText = ''; + if (text.trim()) await this.postBufferedText(thread, text); }; // Don't start streaming post eagerly — wait for first text delta const ensureStreamingPost = () => { + if (streamingStopped) return; if (!streamingPost) startStreamingPost(); }; const responseLifecycle = this.createResponseLifecycle({ @@ -192,7 +220,11 @@ export class AgentChatStreamConsumer { case 'text-delta': { if (responseState.suppressText) break; const { delta } = chunk; + // Opening on whitespace alone posts a placeholder that the + // platform then has nothing to replace it with. + if (!delta.trim() && !responseState.hasVisibleResponse) break; await responseLifecycle.startStreamingResponse(); + if (retainText) pendingText += delta; textStream.yield?.(delta); if (delta.trim()) responseState.hasVisibleResponse = true; break; @@ -219,7 +251,12 @@ export class AgentChatStreamConsumer { break; case 'tool-result': this.noteToolResult(chunk, responseState); - if (this.isSilentOutcome(chunk)) responseState.suppressText = true; + if (this.isSilentOutcome(chunk)) { + responseState.suppressText = true; + // Whatever already reached the platform cannot be recalled, but + // text still pending can be dropped. + pendingText = ''; + } break; default: // Ignore non-user-visible chunks (reasoning, finish, @@ -233,6 +270,26 @@ export class AgentChatStreamConsumer { } } + /** + * `{ markdown }` matches what Chat SDK's streaming path sends, so the adapter + * applies its markdown parse mode. A raw string renders as plain text. + */ + private async postBufferedText( + thread: Thread, + text: string, + throwOnDeliveryError = false, + ): Promise { + try { + await thread.post({ markdown: text }); + } catch (postError: unknown) { + this.options.logger.error('[AgentChatBridge] Buffered post failed', { + error: postError instanceof Error ? postError.message : String(postError), + }); + if (throwOnDeliveryError) throw postError; + await this.options.postErrorToThread(thread, postError); + } + } + private noteToolResult(chunk: ToolResultChunk, state: ResponseState): void { if (isRateLimitedToolOutput(chunk.output)) { state.fallbackSource = 'rate-limit'; @@ -335,21 +392,8 @@ export class AgentChatStreamConsumer { const text = buffer; buffer = ''; if (!text.trim()) return; - try { - await responseLifecycle.startDiscreteResponse(); - // Chat SDK's streaming path wraps accumulated deltas as `{ markdown }` - // so the platform adapter applies its markdown parse-mode (Telegram: - // sendMessage with parse_mode=Markdown). A raw string bypasses that - // and renders as plain text, so we post the buffered message the same - // shape the streaming path uses under the hood. - await thread.post({ markdown: text }); - } catch (postError: unknown) { - this.options.logger.error('[AgentChatBridge] Buffered post failed', { - error: postError instanceof Error ? postError.message : String(postError), - }); - if (options.throwOnDeliveryError) throw postError; - await this.options.postErrorToThread(thread, postError); - } + await responseLifecycle.startDiscreteResponse(); + await this.postBufferedText(thread, text, options.throwOnDeliveryError); responseState.hasVisibleResponse = true; }; diff --git a/packages/cli/src/modules/agents/integrations/chat-integration.service.ts b/packages/cli/src/modules/agents/integrations/chat-integration.service.ts index d2c390674ec..54e65479e38 100644 --- a/packages/cli/src/modules/agents/integrations/chat-integration.service.ts +++ b/packages/cli/src/modules/agents/integrations/chat-integration.service.ts @@ -127,6 +127,26 @@ async function getAgentExecutionOrchestratorService() { return Container.get(AgentExecutionOrchestratorService); } +/** + * Teams renders a reply progressively by editing a posted message, not through + * the platform's own streaming protocol. + * + * That protocol needs a handle the adapter holds only for the life of the + * inbound request, and a turn runs later, from the queue. Giving the SDK a + * placeholder makes the Teams adapter decline to stream and hand back to the + * SDK's post-and-edit path, which uses ordinary proactive calls. + * + * The interval trades smoothness against Teams' edit throttling. The SDK waits + * for each edit to land before scheduling the next, so a throttled tenant paces + * itself, and a skipped interval edit is invisible — but the SDK's final edit is + * not guarded, so a rejection there would surface instead of the last chunk. + * Hence a little above the SDK's 500ms default rather than at it. + */ +function streamingOptionsFor(type: AgentIntegrationConfig['type']) { + if (type !== 'teams') return {}; + return { fallbackStreamingPlaceholderText: '…', streamingUpdateIntervalMs: 750 }; +} + /** * Manages per-agent Chat SDK instances and their lifecycle. * @@ -1192,6 +1212,7 @@ export class ChatIntegrationService { // bot.webhooks.slack maps correctly to the handler. adapters: { [integration.type]: adapter } as Record, state, + ...streamingOptionsFor(integration.type), }); if (ingressEnabled) bridge = await this.createChatBridge(chat, ctx); diff --git a/packages/cli/src/modules/agents/integrations/platforms/__tests__/teams/synthetic-integration.test.ts b/packages/cli/src/modules/agents/integrations/platforms/__tests__/teams/synthetic-integration.test.ts index 81eed5f8c7a..830fe4b84e1 100644 --- a/packages/cli/src/modules/agents/integrations/platforms/__tests__/teams/synthetic-integration.test.ts +++ b/packages/cli/src/modules/agents/integrations/platforms/__tests__/teams/synthetic-integration.test.ts @@ -1,6 +1,10 @@ +import type { StreamChunk } from '@n8n/agents'; import { isRecord } from '@n8n/utils/is-record'; -import { createTeamsReplayContext } from '../../../__tests__/helpers/teams/replay-test-context'; +import { + createTeamsReplayContext, + streamInfo, +} from '../../../__tests__/helpers/teams/replay-test-context'; import { cardAction, channelFollowUp, @@ -69,11 +73,13 @@ describe('Microsoft Teams integration scenarios', () => { integrationType: 'teams', }), ); + // A direct message renders progressively, so the text lands as an edit + // of the placeholder rather than on the first post. expect(ctx.lastPost()?.body).toMatchObject({ type: 'message', - text: 'Got it', conversation: { id: TEAMS_DM_CONVERSATION_ID }, }); + expect(ctx.lastEdit()?.body).toMatchObject({ text: 'Got it' }); } finally { await ctx.shutdown(); } @@ -517,3 +523,160 @@ describe('Microsoft Teams integration scenarios', () => { } }); }); + +const threeDeltas: StreamChunk[] = [ + { type: 'text-delta', id: 't-1', delta: 'Looking ' }, + { type: 'text-delta', id: 't-1', delta: 'into it ' }, + { type: 'text-delta', id: 't-1', delta: 'now' }, + { type: 'finish', finishReason: 'stop' }, +]; + +describe('Microsoft Teams streaming', () => { + it('renders a direct message reply progressively, then settles on the full text', async () => { + // A gap between deltas so the renderer's timer actually ticks; without one + // the whole reply drains before the first interval and only the final edit + // runs, which would not exercise progressive rendering at all. + const ctx = await createTeamsReplayContext({ stream: threeDeltas, streamGapMs: 40 }); + try { + await ctx.sendWebhook(dmMessage); + + const activities = ctx.activities(); + // Teams' own streaming protocol needs a handle that only exists while + // the inbound request is open, so this is post-and-edit instead. + expect(activities.every((call) => streamInfo(call.body) === undefined)).toBe(true); + + const posts = activities.filter((call) => call.body.type === 'message'); + expect(posts).toHaveLength(1); + expect(posts[0].body.text).toBe('…'); + + const edits = ctx.edits().map((call) => String(call.body.text)); + expect(edits.length).toBeGreaterThan(1); + // Each edit carries what came before it, ending on the whole reply. + expect(edits.at(-1)).toBe('Looking into it now'); + expect(edits.at(-1)?.startsWith(edits[0])).toBe(true); + } finally { + await ctx.shutdown(); + } + }); + + it('still delivers the reply when every edit is rejected', async () => { + const ctx = await createTeamsReplayContext({ + stream: threeDeltas, + streamGapMs: 40, + // Nothing reaches the placeholder, so the reply would be lost behind it. + succeedingEdits: 0, + }); + try { + await expect(ctx.sendWebhook(dmMessage)).resolves.toMatchObject({ status: 200 }); + + const texts = ctx + .activities() + .map((call) => call.body.text) + .filter((text): text is string => typeof text === 'string'); + expect(texts).toContain('Looking into it now'); + } finally { + await ctx.shutdown(); + } + }); + + /** + * The known cost of that recovery. The SDK guards its interval edits but not + * its last one, and it reports one promise for all of them, so a turn whose + * earlier edits landed cannot be told apart from one where nothing did. The + * reply is repeated rather than lost, which is the better of the two. + */ + it('repeats a reply whose earlier edits landed before one was rejected', async () => { + const ctx = await createTeamsReplayContext({ + stream: threeDeltas, + streamGapMs: 40, + succeedingEdits: 1, + }); + try { + await ctx.sendWebhook(dmMessage); + + // The edit that landed left part of the reply in the placeholder. + const edits = ctx.edits().map((call) => String(call.body.text)); + expect(edits[0]).not.toBe(''); + expect('Looking into it now'.startsWith(edits[0])).toBe(true); + + // The whole reply then arrives again, as its own message. + const messages = ctx + .activities() + .filter((call) => call.body.type === 'message' && call.body.text !== '…'); + expect(messages).toHaveLength(1); + expect(messages[0].body.text).toBe('Looking into it now'); + } finally { + await ctx.shutdown(); + } + }); + + it('shows a plain typing indicator before the reply starts', async () => { + const ctx = await createTeamsReplayContext({ stream: threeDeltas }); + try { + await ctx.sendWebhook(dmMessage); + + const activities = ctx.activities(); + const indicators = activities.filter( + (call) => call.body.type === 'typing' && !streamInfo(call.body), + ); + expect(indicators).toHaveLength(1); + // It has to arrive before anything else, or it tells the user nothing. + expect(activities.indexOf(indicators[0])).toBe(0); + } finally { + await ctx.shutdown(); + } + }); + + it('sends a channel reply as one message, with no placeholder and no edits', async () => { + const ctx = await createTeamsReplayContext({ stream: threeDeltas }); + try { + await ctx.sendWebhook(channelMention); + + const messages = ctx.activities().filter((call) => call.body.type === 'message'); + expect(messages).toHaveLength(1); + expect(messages[0].body.text).toContain('Looking into it now'); + expect(ctx.edits()).toHaveLength(0); + } finally { + await ctx.shutdown(); + } + }); + + it('posts text after a card as its own message', async () => { + const ctx = await createTeamsReplayContext({ + stream: [ + { type: 'text-delta', id: 't-1', delta: 'Deploying now.' }, + { + type: 'tool-call-suspended', + runId: 'run-order-1', + toolCallId: 'tool-order-1', + toolName: 'approval', + suspendPayload: { + type: 'approval', + toolName: 'send_teams_message', + displayName: 'Send Teams message', + args: { text: 'Continue?' }, + }, + }, + { type: 'text-delta', id: 't-2', delta: 'Waiting on you.' }, + { type: 'finish', finishReason: 'stop' }, + ], + }); + try { + await ctx.sendWebhook(dmMessage); + + // The trailing text is its own message rather than an edit folded back + // into the bubble that sits above the card. + const trailing = ctx + .activities() + .filter( + (call) => + call.body.type === 'message' && + typeof call.body.text === 'string' && + call.body.text.includes('Waiting on you.'), + ); + expect(trailing).toHaveLength(1); + } finally { + await ctx.shutdown(); + } + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/platforms/teams/teams-integration.ts b/packages/cli/src/modules/agents/integrations/platforms/teams/teams-integration.ts index a30ac09b3f0..781f99fed53 100644 --- a/packages/cli/src/modules/agents/integrations/platforms/teams/teams-integration.ts +++ b/packages/cli/src/modules/agents/integrations/platforms/teams/teams-integration.ts @@ -10,17 +10,24 @@ import { AgentChatIntegration, type AgentChannelPreconditionContext, type AgentChatIntegrationContext, + type BridgeExecutionContext, + type BridgeMessageContextParams, + type BridgeResumeExecutionContext, } from '../../agent-chat-integration'; import { expandSelectsToButtons, type SuspendComponent } from '../../component-mapper'; import { assertCredentialNotClaimed } from '../../credential-claim'; import { loadTeamsAdapter } from '../../esm-loader'; import { resolveIntegrationActionDefinitions } from '../../integration-tool-definitions'; +import { startTypingIndicator } from '../typing-indicator'; /** Pinned so a stray TEAMS_API_URL env var cannot redirect proactive sends. */ const TEAMS_API_URL = 'https://smba.trafficmanager.net/teams'; const GLOBAL_GRAPH_API_BASE_URL = 'https://graph.microsoft.com'; +/** Interval picked to match Discord's; Teams does not document the expiry. */ +const TEAMS_TYPING_REFRESH_MS = 8000; + /** * A tenant ID is a GUID or a verified domain. The value reaches the Teams SDK, * which interpolates it into a token URL path, so the shape is checked before @@ -109,7 +116,18 @@ export class TeamsIntegration extends AgentChatIntegration { */ readonly targetSuspensionCardAtActingUser = true; - readonly disableStreaming = true; + /** + * A direct message renders progressively; every other conversation posts one + * message. The choice is made for each conversation in + * `createBridgeExecutionContext` rather than here. + */ + readonly disableStreaming = false; + + /** + * Text that follows a card is posted on its own rather than folded back into + * the message being edited above it. + */ + readonly singleStreamedRunPerTurn = true; constructor( private readonly logger: Logger, @@ -156,6 +174,52 @@ export class TeamsIntegration extends AgentChatIntegration { return expandSelectsToButtons(components); } + /** + * Only a direct message renders progressively. Post-and-edit would work in a + * channel too, but a message that visibly rewrites itself is far more + * disruptive there, so that stays a separate decision. + */ + async createBridgeExecutionContext( + params: BridgeMessageContextParams, + ): Promise { + const streamable = params.thread.isDM; + return { + platformAgentContext: {}, + forceBuffered: !streamable, + // A queued message is only captured here; the turn that would clear the + // indicator runs later, so starting one now leaves it refreshing alone. + statusHandle: + params.startStatus === false + ? undefined + : this.startTyping(params.thread, params.logger, params.agentId), + }; + } + + /** A card action arrives as an invoke activity, which carries no streamer. */ + async createResumeExecutionContext(params: { + thread: BridgeMessageContextParams['thread']; + logger: BridgeMessageContextParams['logger']; + agentId: string; + }): Promise { + return { + forceBuffered: true, + statusHandle: this.startTyping(params.thread, params.logger, params.agentId), + }; + } + + private startTyping( + thread: BridgeMessageContextParams['thread'], + logger: BridgeMessageContextParams['logger'], + agentId: string, + ) { + return startTypingIndicator(thread, { + logger, + agentId, + platform: 'Microsoft Teams', + refreshMs: TEAMS_TYPING_REFRESH_MS, + }); + } + /** * The certificate check is ours to make: that mode stores no `clientSecret`, * so passing it through would build an adapter with no secret and fail later