feat(core): Render Microsoft Teams agent replies progressively (no-changelog) (#39514)

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, unknown>)?.[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<string>) {
+			// Drain the iterable the way a real adapter does.
+		}
+		return undefined;
+	});
+	const thread = mock<Thread<unknown, unknown>>();
+	thread.post = post as unknown as typeof thread.post;
+	return { thread, streamed, discrete };
+}
+
+function makeStreamingConsumer(
+	options: Partial<ConstructorParameters<typeof AgentChatStreamConsumer>[0]> = {},
+) {
+	return new AgentChatStreamConsumer({
+		disableStreaming: false,
+		logger: mock<Logger>(),
+		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');

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: {

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<StreamChunk> {
+export function toStream(chunks: StreamChunk[], gapMs = 0): AsyncGenerator<StreamChunk> {
 	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<TChat extends ChatInstance>(params: {
 	integration: AgentIntegrationConfig;
 	componentMapper?: ComponentMapper;
 	stream?: StreamChunk[];
+	/** Delay between chunks, so a timer-driven renderer ticks. */
+	streamGapMs?: number;
 }): ReplayContextSetup<TChat> {
 	const registry = new ChatIntegrationRegistry();
 	registry.register(params.integrationImpl);
@@ -226,11 +232,11 @@ export function createReplayContextSetup<TChat extends ChatInstance>(params: {
 		executeForChatPublished: vi.fn<AgentExecutor['executeForChatPublished']>((config) => {
 			selectedContext = config.messageContext ?? undefined;
 			selectedThreadId = config.memory.threadId.id;
-			return toStream(stream);
+			return toStream(stream, params.streamGapMs);
 		}),
 		resumeForChat: vi.fn<AgentExecutor['resumeForChat']>((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.

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<ReplayContextSetup, 'chat'> {
 	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<string, unknown>): Record<string, unknown> | undefined {
+	const entities = body.entities;
+	if (!Array.isArray(entities)) return undefined;
+	return entities.find((entity) => isRecord(entity) && entity.type === 'streaminfo') as
+		| Record<string, unknown>
+		| 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<string, unknown>,
-			});
+			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<string, unknown>,
 			});
+			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<TeamsReplayContext> {
 	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<string, never>,
 		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<BackendLogger>(), mock<AgentRepository>());
 	const setup = createReplayContextSetup({
 		chat: chat as never,
-		integrationImpl: new TeamsIntegration(mock<BackendLogger>(), mock<AgentRepository>()),
+		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 () => {

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<Logger>(),
+				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();
+		});
+	});
 });

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),

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

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<unknown> | 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<string> => {
 			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<unknown, unknown>,
+		text: string,
+		throwOnDeliveryError = false,
+	): Promise<void> {
+		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;
 		};
 

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<string, never>,
 				state,
+				...streamingOptionsFor(integration.type),
 			});
 
 			if (ingressEnabled) bridge = await this.createChatBridge(chat, ctx);

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();
+		}
+	});
+});

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<BridgeExecutionContext> {
+		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<BridgeResumeExecutionContext> {
+		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