From e64291dc6d2bf03c0ca90fbd65e1d862e46da43c Mon Sep 17 00:00:00 2001 From: Bernhard Wittmann Date: Fri, 25 Sep 2026 10:13:14 +0000 Subject: [PATCH] fix(core): Make a Microsoft Teams approval card private and self-clearing (no-changelog) (#39515) --- .../agent-conversation-state.service.test.ts | 61 +++ ...ent-execution-orchestrator.service.test.ts | 8 +- .../agent-conversation-state.service.ts | 10 + .../agent-execution-orchestrator.service.ts | 5 +- .../agent-chat-hitl-resume-handler.test.ts | 384 +++++++++++++++--- .../agent-chat-stream-consumer.test.ts | 41 ++ .../__tests__/callback-store.test.ts | 4 +- .../helpers/linear/replay-test-context.ts | 1 + .../__tests__/helpers/replay-test-helpers.ts | 4 + .../helpers/teams/replay-test-context.ts | 10 + .../helpers/teams/synthetic-fixtures.ts | 41 +- .../helpers/telegram/replay-test-context.ts | 1 + .../__tests__/teams-integration.test.ts | 49 +-- .../agents/integrations/agent-chat-bridge.ts | 60 ++- .../agent-chat-hitl-resume-handler.ts | 104 ++++- .../integrations/agent-chat-integration.ts | 32 ++ .../agent-chat-stream-consumer.ts | 10 +- .../agents/integrations/callback-store.ts | 3 +- .../teams/synthetic-integration.test.ts | 158 ++++++- .../platforms/teams/teams-integration.ts | 25 +- patches/@chat-adapter__teams@4.37.0.patch | 130 ++++++ pnpm-lock.yaml | 5 +- pnpm-workspace.yaml | 1 + 23 files changed, 987 insertions(+), 160 deletions(-) create mode 100644 packages/cli/src/modules/agents/__tests__/agent-conversation-state.service.test.ts create mode 100644 patches/@chat-adapter__teams@4.37.0.patch diff --git a/packages/cli/src/modules/agents/__tests__/agent-conversation-state.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-conversation-state.service.test.ts new file mode 100644 index 00000000000..e187fda4ad1 --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/agent-conversation-state.service.test.ts @@ -0,0 +1,61 @@ +import type { SerializableAgentState } from '@n8n/agents'; +import { mock } from 'vitest-mock-extended'; + +import { AgentConversationStateService } from '../agent-conversation-state.service'; +import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; +import type { AgentExecutionRepository } from '../repositories/agent-execution.repository'; + +const AGENT_ID = 'agent-1'; +const RUN_ID = 'run-1'; + +function checkpoint(status: SerializableAgentState['status']): SerializableAgentState { + return mock({ status }); +} + +describe('AgentConversationStateService', () => { + describe('isResumable', () => { + const executionRepository = mock(); + + function createService(status: Awaited>) { + const checkpointStorage = mock({ + getStatus: vi.fn(async () => status), + }); + const service = new AgentConversationStateService(executionRepository, checkpointStorage); + return { service, checkpointStorage }; + } + + it('accepts a checkpoint that is still suspended', async () => { + const { service, checkpointStorage } = createService({ + status: 'active', + checkpoint: checkpoint('suspended'), + }); + + await expect(service.isResumable(AGENT_ID, RUN_ID)).resolves.toBe(true); + expect(checkpointStorage.getStatus).toHaveBeenCalledWith(RUN_ID, AGENT_ID); + }); + + it.each(['running', 'success', 'failed', 'waiting', 'idle'] as const)( + 'rejects a checkpoint the runtime already took (%s)', + async (status) => { + const { service } = createService({ status: 'active', checkpoint: checkpoint(status) }); + + await expect(service.isResumable(AGENT_ID, RUN_ID)).resolves.toBe(false); + }, + ); + + it('rejects an expired checkpoint', async () => { + const { service } = createService({ + status: 'expired', + checkpoint: checkpoint('suspended'), + }); + + await expect(service.isResumable(AGENT_ID, RUN_ID)).resolves.toBe(false); + }); + + it('rejects a checkpoint that is gone', async () => { + const { service } = createService({ status: 'not-found' }); + + await expect(service.isResumable(AGENT_ID, RUN_ID)).resolves.toBe(false); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index 93752601215..b329b98172e 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -2032,7 +2032,13 @@ describe('AgentExecutionOrchestratorService', () => { }); expect(externalHooks.run).toHaveBeenCalledWith('agent.preExecute', [agentId]); expect(chatIntegrationService.getBridge).toHaveBeenCalledWith(agentId, 'slack', 'credential-1'); - expect(bridge.deliverWakeResponse).toHaveBeenCalledWith('slack:channel-1:1', chunks); + // The third argument keeps a card the woken turn raises addressed to the + // person the conversation belongs to, on platforms that can target one. + expect(bridge.deliverWakeResponse).toHaveBeenCalledWith( + 'slack:channel-1:1', + chunks, + 'selected-user', + ); expect( readIntegrationMessageContext(runtime.agent.stream.mock.calls[0][1].persistence), ).toEqual(messageContext); diff --git a/packages/cli/src/modules/agents/agent-conversation-state.service.ts b/packages/cli/src/modules/agents/agent-conversation-state.service.ts index 93d3d90f25a..ed3e1a99d54 100644 --- a/packages/cli/src/modules/agents/agent-conversation-state.service.ts +++ b/packages/cli/src/modules/agents/agent-conversation-state.service.ts @@ -22,4 +22,14 @@ export class AgentConversationStateService { ]); return { running, suspendedCheckpoint }; } + + /** + * A click on a card whose run expired or was already resolved must not resume + * it. The inner state matters too: `active` only says the row is present and + * within its TTL, while the runtime rejects anything it has already claimed. + */ + async isResumable(agentId: string, runId: string): Promise { + const result = await this.checkpointStorage.getStatus(runId, agentId); + return result.status === 'active' && result.checkpoint.status === 'suspended'; + } } diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index 427a808a075..7988a8c2f03 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -652,6 +652,7 @@ export class AgentExecutionOrchestratorService { this.streamWakeTurn(config, runtime, access, integrationType, messageContext), abortSignal, delivery, + messageContext?.interactingUserId, ), ); for await (const _chunk of stream) { @@ -663,6 +664,7 @@ export class AgentExecutionOrchestratorService { stream: AsyncGenerator, abortSignal: AbortSignal, delivery?: { bridge: AgentChatBridge; threadId: string }, + cardRecipientId?: string, ): AsyncGenerator { const chunks: StreamChunk[] = []; let runError: unknown; @@ -677,7 +679,8 @@ export class AgentExecutionOrchestratorService { throw new OperationalError('Background job wake failed', { cause: runError }); } abortSignal.throwIfAborted(); - if (delivery) await delivery.bridge.deliverWakeResponse(delivery.threadId, chunks); + if (delivery) + await delivery.bridge.deliverWakeResponse(delivery.threadId, chunks, cardRecipientId); } private async getWakeDelivery( diff --git a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts index 6b68f1841ce..98c79f6ab70 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts @@ -1,71 +1,345 @@ +import type { Author } from 'chat'; + import { AgentChatHitlResumeHandler } from '../agent-chat-hitl-resume-handler'; -it.each([ - { - name: 'approval', - callback: { - actionId: 'resume:run-1:tool-1:0', - value: JSON.stringify({ approved: true }), - kind: 'approval' as const, - }, - content: '✅ Approved by Alice', - }, - { - name: 'non-approval selection', - callback: { - actionId: 'resume:run-1:tool-1:0', - value: JSON.stringify({ type: 'button', value: 'continue' }), - label: 'Continue', - }, - content: '✅ Continue selected by Alice', - }, -])('settles a $name card in place before resuming the agent', async ({ callback, content }) => { - const settleActionMessage = vi.fn().mockResolvedValue(undefined); - const deleteMessage = vi.fn().mockResolvedValue(undefined); - const resumeForChat = vi.fn(() => (async function* () {})()); - const handler = new AgentChatHitlResumeHandler({ +const THREAD_ID = 'discord:800000000000000001:700000000000000001:600000000000000001'; +const ALICE: Author = { + userId: 'user-1', + userName: 'alice', + fullName: 'Alice', +} as Author; + +type HandlerOptions = ConstructorParameters[0]; + +function createHandler(overrides: Partial = {}) { + return new AgentChatHitlResumeHandler({ agentId: 'agent-1', projectId: 'project-1', integration: { type: 'discord', credentialId: 'cred-1' }, - agentService: { resumeForChat }, + agentService: { resumeForChat: vi.fn(() => (async function* () {})()) }, logger: { warn: vi.fn() } as never, - callbackStore: { - resolve: vi.fn().mockResolvedValue(callback), - } as never, deleteActionMessageBeforeResume: false, - formatActionDecisionMessage: ({ approved, selectedLabel, user }) => - approved === true - ? `✅ Approved by ${user.fullName}` - : `✅ ${selectedLabel} selected by ${user.fullName}`, - settleActionMessage, - resolvePlatformThreadId: () => - 'discord:800000000000000001:700000000000000001:600000000000000001', + resolvePlatformThreadId: () => THREAD_ID, toAgentThreadId: () => ({ id: 'agent-thread-1' }) as never, getPlatformAgentContext: () => ({}), messageContextBridge: { capture: vi.fn().mockReturnValue(undefined) } as never, streamConsumer: { consume: vi.fn().mockResolvedValue(undefined) } as never, createResumeExecutionContext: async () => ({}), - }); + ...overrides, + } as HandlerOptions); +} - await handler.handleAction({ - actionId: 'callback-key', - thread: { post: vi.fn() }, - threadId: 'discord:800000000000000001:700000000000000001:600000000000000001', - messageId: 'message-1', - user: { userId: 'user-1', userName: 'alice', fullName: 'Alice' }, - adapter: { deleteMessage }, - raw: {}, - } as never); +function createThread(nativeEphemeral: boolean) { + return { + post: vi.fn().mockResolvedValue(undefined), + postEphemeral: vi + .fn() + .mockResolvedValue(nativeEphemeral ? { id: 'ephemeral-1', usedFallback: false } : null), + }; +} - expect(deleteMessage).not.toHaveBeenCalled(); - expect(settleActionMessage).toHaveBeenCalledWith({ - agentId: 'agent-1', - integration: { type: 'discord', credentialId: 'cred-1' }, - threadId: 'discord:800000000000000001:700000000000000001:600000000000000001', - messageId: 'message-1', - content, - }); - expect(settleActionMessage.mock.invocationCallOrder[0]).toBeLessThan( - resumeForChat.mock.invocationCallOrder[0], +describe('settling the action card', () => { + it.each([ + { + name: 'approval', + callback: { + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + }, + content: '✅ Approved by Alice', + }, + { + name: 'non-approval selection', + callback: { + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ type: 'button', value: 'continue' }), + label: 'Continue', + }, + content: '✅ Continue selected by Alice', + }, + // Teams and any other platform that carries the full payload in the card: + // the decision is in the button value, so no store is needed to name it. + { + name: 'store-less approval', + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + content: '✅ Approved by Alice', + }, + { + name: 'store-less decline', + actionId: 'resume:run-1:tool-1:1', + value: JSON.stringify({ approved: false }), + content: '🚫 Declined by Alice', + }, + ])( + 'names the decision on a $name card before resuming the agent', + async ({ callback, actionId = 'callback-key', value, content }) => { + const settleActionMessage = vi.fn().mockResolvedValue(undefined); + const deleteMessage = vi.fn().mockResolvedValue(undefined); + const resumeForChat = vi.fn(() => (async function* () {})()); + const handler = createHandler({ + agentService: { resumeForChat }, + ...(callback + ? { callbackStore: { resolve: vi.fn().mockResolvedValue(callback) } as never } + : {}), + formatActionDecisionMessage: ({ approved, selectedLabel, user }) => { + if (approved === undefined) return `✅ ${selectedLabel} selected by ${user.fullName}`; + return approved ? `✅ Approved by ${user.fullName}` : `🚫 Declined by ${user.fullName}`; + }, + settleActionMessage, + }); + + await handler.handleAction({ + actionId, + value, + thread: createThread(true), + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage }, + raw: {}, + } as never); + + expect(deleteMessage).not.toHaveBeenCalled(); + expect(settleActionMessage).toHaveBeenCalledWith({ + agentId: 'agent-1', + integration: { type: 'discord', credentialId: 'cred-1' }, + threadId: THREAD_ID, + messageId: 'message-1', + content, + }); + expect(settleActionMessage.mock.invocationCallOrder[0]).toBeLessThan( + resumeForChat.mock.invocationCallOrder[0], + ); + }, ); }); + +describe('notices that answer one click', () => { + const EXPIRED_NOTICE = + 'This action is no longer available. The link may have expired or already been used.'; + + it('tells only the clicking user that their callback key expired', async () => { + const thread = createThread(true); + const resumeForChat = vi.fn(() => (async function* () {})()); + const handler = createHandler({ + agentService: { resumeForChat }, + callbackStore: { resolve: vi.fn().mockResolvedValue(undefined) } as never, + }); + + await handler.handleAction({ + actionId: 'callback-key', + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage: vi.fn() }, + raw: {}, + } as never); + + expect(thread.postEphemeral).toHaveBeenCalledWith(ALICE, EXPIRED_NOTICE, { + fallbackToDM: false, + }); + expect(thread.post).not.toHaveBeenCalled(); + expect(resumeForChat).not.toHaveBeenCalled(); + }); + + it('falls back to the thread when the ephemeral post is rejected', async () => { + const thread = { + post: vi.fn().mockResolvedValue(undefined), + postEphemeral: vi.fn().mockRejectedValue(new Error('rate limited')), + }; + const handler = createHandler({ + callbackStore: { resolve: vi.fn().mockResolvedValue(undefined) } as never, + }); + + await handler.handleAction({ + actionId: 'callback-key', + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage: vi.fn() }, + raw: {}, + } as never); + + expect(thread.post).toHaveBeenCalledWith(EXPIRED_NOTICE); + }); + + it('falls back to the thread where the platform has no ephemeral message', async () => { + const thread = createThread(false); + const handler = createHandler({ + callbackStore: { resolve: vi.fn().mockResolvedValue(undefined) } as never, + }); + + await handler.handleAction({ + actionId: 'callback-key', + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage: vi.fn() }, + raw: {}, + } as never); + + expect(thread.post).toHaveBeenCalledWith(EXPIRED_NOTICE); + }); + + it('tells only the clicking user that the action was already handled', async () => { + const thread = createThread(true); + // Hold the first resume open so the second one meets the in-flight guard. + let releaseFirstResume!: () => void; + const firstResumeHeld = new Promise((resolve) => { + releaseFirstResume = resolve; + }); + const handler = createHandler({ + streamConsumer: { consume: async () => await firstResumeHeld } as never, + }); + + const first = handler.executeResume( + thread as never, + 'run-1', + 'tool-1', + {}, + { + actingUser: ALICE, + }, + ); + await handler.executeResume(thread as never, 'run-1', 'tool-1', {}, { actingUser: ALICE }); + releaseFirstResume(); + await first; + + expect(thread.postEphemeral).toHaveBeenCalledWith( + ALICE, + 'This action has already been handled', + { + fallbackToDM: false, + }, + ); + expect(thread.post).not.toHaveBeenCalled(); + }); + + it('stays silent on a resume no user triggered', async () => { + const thread = createThread(true); + let releaseFirstResume!: () => void; + const firstResumeHeld = new Promise((resolve) => { + releaseFirstResume = resolve; + }); + const handler = createHandler({ + streamConsumer: { consume: async () => await firstResumeHeld } as never, + }); + + const first = handler.executeResume(thread as never, 'run-1', 'tool-1', {}); + await handler.executeResume(thread as never, 'run-1', 'tool-1', {}); + releaseFirstResume(); + await first; + + expect(thread.postEphemeral).not.toHaveBeenCalled(); + expect(thread.post).not.toHaveBeenCalled(); + }); + + it('tells the clicking user when the run behind the card is gone', async () => { + const thread = createThread(true); + const resumeForChat = vi.fn(() => (async function* () {})()); + const settleActionMessage = vi.fn().mockResolvedValue(undefined); + const handler = createHandler({ + agentService: { resumeForChat, isResumable: async () => false }, + formatActionDecisionMessage: () => 'never settled', + settleActionMessage, + }); + + await handler.handleAction({ + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage: vi.fn() }, + raw: {}, + } as never); + + expect(thread.postEphemeral).toHaveBeenCalledWith(ALICE, EXPIRED_NOTICE, { + fallbackToDM: false, + }); + // No resume, so the misconfiguration error never reaches the thread. + expect(resumeForChat).not.toHaveBeenCalled(); + expect(settleActionMessage).not.toHaveBeenCalled(); + }); + + it('removes a stale card on a platform that deletes answered ones', async () => { + const thread = createThread(true); + const deleteMessage = vi.fn().mockResolvedValue(undefined); + const handler = createHandler({ + agentService: { + resumeForChat: vi.fn(() => (async function* () {})()), + isResumable: async () => false, + }, + deleteActionMessageBeforeResume: true, + }); + + await handler.handleAction({ + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage }, + raw: {}, + } as never); + + expect(deleteMessage).toHaveBeenCalledWith(THREAD_ID, 'message-1'); + }); + + it('resumes when the run is still resumable', async () => { + const thread = createThread(true); + const resumeForChat = vi.fn(() => (async function* () {})()); + const handler = createHandler({ + agentService: { resumeForChat, isResumable: async () => true }, + }); + + await handler.handleAction({ + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage: vi.fn() }, + raw: {}, + } as never); + + expect(resumeForChat).toHaveBeenCalled(); + expect(thread.postEphemeral).not.toHaveBeenCalled(); + }); + + it('removes an answered card on a platform that deletes them, then resumes', async () => { + const thread = createThread(true); + const resumeForChat = vi.fn(() => (async function* () {})()); + const deleteMessage = vi.fn().mockResolvedValue(undefined); + const settleActionMessage = vi.fn(); + const handler = createHandler({ + agentService: { resumeForChat, isResumable: async () => true }, + deleteActionMessageBeforeResume: true, + settleActionMessage, + }); + + await handler.handleAction({ + actionId: 'resume:run-1:tool-1:0', + value: JSON.stringify({ approved: true }), + thread, + threadId: THREAD_ID, + messageId: 'message-1', + user: ALICE, + adapter: { deleteMessage }, + raw: {}, + } as never); + + expect(deleteMessage).toHaveBeenCalledWith(THREAD_ID, 'message-1'); + expect(settleActionMessage).not.toHaveBeenCalled(); + expect(resumeForChat).toHaveBeenCalledWith( + expect.objectContaining({ runId: 'run-1', resumeData: { approved: true } }), + ); + }); +}); 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 a21c38925f0..d5296a6b1a0 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 @@ -27,6 +27,13 @@ function makeConsumer( const thread = mock>(); +const SUSPENDED_CHUNK = { + type: 'tool-call-suspended', + runId: 'run-1', + toolCallId: 'tool-1', + suspendPayload: {}, +} as unknown as StreamChunk; + describe('AgentChatStreamConsumer — rate-limit fallback', () => { it('posts a fallback error after a RATE_LIMIT_EXCEEDED tool result', async () => { const postErrorToThread = vi.fn().mockResolvedValue(undefined); @@ -112,3 +119,37 @@ describe('AgentChatStreamConsumer — rate-limit fallback', () => { expect(postErrorToThread).not.toHaveBeenCalled(); }); }); + +describe('AgentChatStreamConsumer — suspension routing', () => { + function makeSuspensionConsumer(disableStreaming: boolean) { + const handleSuspension = vi.fn().mockResolvedValue('posted'); + const consumer = new AgentChatStreamConsumer({ + disableStreaming, + logger: mock(), + postErrorToThread: vi.fn().mockResolvedValue(undefined), + handleSuspension, + handleMessage: vi.fn().mockResolvedValue(false), + isIntegrationActionTool: () => true, + }); + return { consumer, handleSuspension }; + } + + it.each([ + ['streaming', false], + ['buffered', true], + ])('passes the acting user to the suspension handler (%s)', async (_name, disableStreaming) => { + const { consumer, handleSuspension } = makeSuspensionConsumer(disableStreaming); + + await consumer.consume(makeStream([SUSPENDED_CHUNK]), thread, { actingUserId: 'user-1' }); + + expect(handleSuspension).toHaveBeenCalledWith(SUSPENDED_CHUNK, thread, 'user-1'); + }); + + it('leaves the acting user unset on a turn no user drove', async () => { + const { consumer, handleSuspension } = makeSuspensionConsumer(true); + + await consumer.consume(makeStream([SUSPENDED_CHUNK]), thread); + + expect(handleSuspension).toHaveBeenCalledWith(SUSPENDED_CHUNK, thread, undefined); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/callback-store.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/callback-store.test.ts index 0bb459c9cc9..7f0074dda3e 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/callback-store.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/callback-store.test.ts @@ -34,13 +34,13 @@ describe('CallbackStore', () => { const reader = new CallbackStore(cache, lockService, 'agent-1:discord:cred-1'); const key = await writer.store('resume:run:tool:0', '{"approved":true}', { - kind: 'approval', + groupId: '["run","tool"]', }); await expect(reader.resolve(key)).resolves.toEqual({ actionId: 'resume:run:tool:0', value: '{"approved":true}', - kind: 'approval', + groupId: '["run","tool"]', }); }); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts index ec6cc4bcc69..8be15b370ba 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts @@ -99,6 +99,7 @@ export interface LinearReplayContext extends Omit[number]; 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 c92f66c20ea..79ba2e2de9a 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 @@ -189,6 +189,7 @@ export interface ReplayContextSetup { agentExecutor: { executeForChatPublished: Mock; resumeForChat: Mock; + isResumable: Mock; }; actionExecutor: ChatIntegrationActionExecutor; descriptor: ReturnType[number]; @@ -231,6 +232,9 @@ export function createReplayContextSetup(params: { selectedContext = config.messageContext ?? undefined; return toStream(stream); }), + // Mirrors how production wires the gate. It admits every run by default, + // so a test that wants the stale branch resolves it to false. + isResumable: vi.fn(async () => true), }; const messageContextStore = new MemoryMessageContextStore(); const pending: Array<{ payload: QueuedIntegrationMessage; threadId: string }> = []; 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 6a3106d8415..91b17c13dba 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 @@ -42,6 +42,7 @@ export interface TeamsReplayContext extends Omit { latestThreadId: () => string | undefined; lastPost: () => ReplayApiCall | undefined; lastEdit: () => ReplayApiCall | undefined; + lastDelete: () => ReplayApiCall | undefined; lastPostedMessageId: () => string | undefined; } @@ -121,6 +122,14 @@ function installTeamsApiStub(jwks: object, accessToken: string) { return [200, { id }]; }); + nock(serviceUrl.origin) + .persist() + .delete(/\/v3\/conversations\/.+\/activities\/.+/) + .reply(function (uri) { + apiCalls.push({ method: 'deleteActivity', body: { uri } }); + return [200, {}]; + }); + nock(serviceUrl.origin) .persist() .put(/\/v3\/conversations\/.+\/activities\/.+/) @@ -196,6 +205,7 @@ export async function createTeamsReplayContext( latestThreadId: setup.latestThreadId, lastPost: () => lastCall('sendActivity'), lastEdit: () => lastCall('updateActivity'), + lastDelete: () => lastCall('deleteActivity'), lastPostedMessageId: () => stub.postedMessageIds.at(-1), shutdown: async () => { stub.restore(); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/synthetic-fixtures.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/synthetic-fixtures.ts index 9f27104a355..047b5139ed2 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/synthetic-fixtures.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/teams/synthetic-fixtures.ts @@ -66,11 +66,50 @@ export const selfMessage: TeamsActivityFixture = baseActivity({ /** * An Adaptive Card button click. Teams delivers `Action.Submit` as a message * activity carrying `value` and no text. + * + * The click is derived from the activity that produced the card, so the + * conversation and its `channelData` cannot drift apart: Teams never sends a + * group chat conversation with the `channelData` of a direct message. */ -export function cardAction(value: Record, replyToId: string) { +export function cardAction( + value: Record, + replyToId: string, + from: TeamsActivityFixture = baseActivity(), +) { return baseActivity({ id: 'activity-action-1', + conversation: from.conversation, + channelData: from.channelData, replyToId, value, }); } + +const mentionEntity = { + type: 'mention', + text: 'n8n Agent', + mentioned: { id: `28:${TEAMS_APP_ID}`, name: 'n8n Agent' }, +}; + +// A group chat conversation id carries no message id, so the whole chat is one +// conversation and one session. + +export const TEAMS_GROUP_CHAT_CONVERSATION_ID = '19:group_chat_test@thread.v2'; + +function groupChatActivity(overrides: Partial = {}): TeamsActivityFixture { + return baseActivity({ + conversation: { + id: TEAMS_GROUP_CHAT_CONVERSATION_ID, + conversationType: 'groupChat', + tenantId: TEAMS_TENANT_ID, + isGroup: true, + }, + ...overrides, + }); +} + +export const groupChatMention: TeamsActivityFixture = groupChatActivity({ + id: 'activity-group-1', + text: 'n8n Agent hello agent', + entities: [mentionEntity], +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts index 5db93da15bb..6a5cd436b8d 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts @@ -92,6 +92,7 @@ export interface TelegramReplayContext extends Omit { }); describe('platform capabilities', () => { - it('reuses the Entra credential, stays hidden, buffers, and settles in place', () => { + it('reuses the Entra credential, stays hidden, buffers, and targets its cards', () => { expect(integration).toMatchObject({ credentialTypes: ['microsoftEntraServicePrincipalApi'], internal: true, @@ -71,7 +71,9 @@ describe('TeamsIntegration', () => { // Left at the base-class default: full Adaptive Card payloads fit, so // no callback store is needed. needsShortCallbackData: false, - deleteActionMessageBeforeResume: false, + targetSuspensionCardAtActingUser: true, + // A targeted card cannot be edited, so an answered one is deleted. + deleteActionMessageBeforeResume: true, }); }); @@ -205,49 +207,6 @@ describe('TeamsIntegration', () => { }); }); - describe('formatActionDecisionMessage', () => { - const user = { userId: 'u-1', userName: 'alice', fullName: 'Alice', isBot: false, isMe: false }; - - it.each([ - [true, '✅ Approved by Alice'], - [false, '🚫 Declined by Alice'], - ])('names the decision when one was resolved: %p', (approved, expected) => { - expect( - integration.formatActionDecisionMessage({ - approved, - raw: {}, - user, - } as Parameters[0]), - ).toBe(expected); - }); - - it.each([ - [{ userId: 'u-1', userName: 'alice', fullName: 'Alice' }, 'Alice'], - [{ userId: 'u-1', userName: 'alice', fullName: '' }, 'alice'], - [{ userId: 'u-1', userName: '', fullName: '' }, 'u-1'], - ])('resolves the responder from %p', (responder, expected) => { - expect( - integration.formatActionDecisionMessage({ - approved: true, - raw: {}, - user: { ...responder, isBot: false, isMe: false }, - } as Parameters[0]), - ).toBe(`✅ Approved by ${expected}`); - }); - - // Without the CallbackStore the resume handler resolves neither the - // decision nor the label, so this generic wording is what a Teams - // approval actually settles to today. - it('falls back to a generic outcome when no decision reaches it', () => { - expect( - integration.formatActionDecisionMessage({ - raw: {}, - user, - } as Parameters[0]), - ).toBe('✅ Action selected by Alice'); - }); - }); - describe('normalizeComponents', () => { it('converts select options into individual buttons', () => { expect( 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 ab9fd175cfd..4e4685ffeab 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts @@ -11,7 +11,7 @@ import { LockNamespace, LockService } from '@n8n/backend-common'; import { type HttpRequestClient, OutboundHttp } from '@n8n/backend-network'; import { Container } from '@n8n/di'; import { isRecord } from '@n8n/utils/is-record'; -import type { Attachment, Author, Chat, Message, Thread } from 'chat'; +import type { Attachment, Author, CardElement, Chat, Message, Thread } from 'chat'; import { UserError, type Logger } from 'n8n-workflow'; import { CacheService } from '@/services/cache/cache.service'; @@ -38,14 +38,18 @@ import type { BridgeExecutionContext, PlatformAgentContext, } from './agent-chat-integration'; -import { ChatIntegrationRegistry, onceStatusHandle } from './agent-chat-integration'; +import { + ChatIntegrationRegistry, + onceStatusHandle, + postToUserOrThread, +} from './agent-chat-integration'; import { AgentChatHitlResumeHandler } from './agent-chat-hitl-resume-handler'; import { AgentChatMessageContextBridge } from './agent-chat-message-context'; import { AgentChatStreamConsumer, type SuspensionHandlingResult, } from './agent-chat-stream-consumer'; -import { buildSuspendCardPayload, isApprovalSuspendPayload } from './agent-chat-suspension-cards'; +import { buildSuspendCardPayload } from './agent-chat-suspension-cards'; import { CallbackStore, type CallbackMetadata } from './callback-store'; import type { ComponentMapper, ShortenCallback } from './component-mapper'; import { loadChatSdk } from './esm-loader'; @@ -142,6 +146,9 @@ interface AgentExecutor extends Pick; + + /** Optional: a caller that cannot look checkpoints up simply skips the gate. */ + isResumable?(config: { agentId: string; runId: string }): Promise; } /** An open checkpoint prevents automatic session rotation. */ @@ -345,6 +352,9 @@ export class AgentChatBridge { ); return suspended ? { suspendPayload: suspended.suspendPayload } : null; }, + async isResumable({ agentId: aid, runId }) { + return await Container.get(AgentConversationStateService).isResumable(aid, runId); + }, }; return new AgentChatBridge( chat, @@ -480,17 +490,32 @@ export class AgentChatBridge { runId, toolCallId, resumeData, - { notifyOnDuplicate: false, ...(context ? { messageContext: context.messageContext } : {}) }, + { + ...(context ? { messageContext: context.messageContext } : {}), + // Nobody clicked, so this must not speak to them. It only keeps a + // card the resumed turn raises addressed to the person whose + // conversation it belongs to. + ...(context?.messageContext?.interactingUserId + ? { cardRecipientId: context.messageContext.interactingUserId } + : {}), + }, ); } - async deliverWakeResponse(threadId: string, chunks: StreamChunk[]): Promise { + async deliverWakeResponse( + threadId: string, + chunks: StreamChunk[], + cardRecipientId?: string, + ): Promise { await this.streamConsumer.consume( (async function* () { yield* chunks; })(), this.chat.thread(threadId), - { throwOnDeliveryError: true }, + { + throwOnDeliveryError: true, + ...(cardRecipientId ? { actingUserId: cardRecipientId } : {}), + }, ); } @@ -847,6 +872,7 @@ export class AgentChatBridge { await this.streamConsumer.consume(stream, thread, { forceBuffered: payload.forceBuffered, statusHandle, + actingUserId: payload.sender.userId, }); } catch (error) { await this.postErrorToThread(thread, error); @@ -1006,6 +1032,7 @@ export class AgentChatBridge { private async handleSuspension( chunk: Extract, thread: Thread, + actingUserId?: string, ): Promise { const { runId, toolCallId, suspendPayload } = chunk; @@ -1018,7 +1045,6 @@ export class AgentChatBridge { if (!cardPayload) return 'skipped'; const callbackMetadata: CallbackMetadata = { groupId: JSON.stringify([runId, toolCallId]), - ...(isApprovalSuspendPayload(suspendPayload) ? { kind: 'approval' } : {}), }; try { @@ -1030,7 +1056,7 @@ export class AgentChatBridge { this.getShortenCallback(callbackMetadata), this.integration.type, ); - await thread.post({ card }); + await this.deliverSuspensionCard(thread, card, actingUserId); return 'posted'; } catch (error) { this.logger.error('[AgentChatBridge] Failed to post suspension card', { @@ -1043,6 +1069,24 @@ export class AgentChatBridge { } } + /** + * Where the platform can address one user, the card goes to them alone, so + * the rest of the conversation never sees it. Everything else falls back to + * the thread: a card nobody receives leaves the run parked with nothing to + * click. + */ + private async deliverSuspensionCard( + thread: Thread, + card: CardElement, + actingUserId?: string, + ): Promise { + if (this.integrationImpl?.targetSuspensionCardAtActingUser && actingUserId) { + await postToUserOrThread(thread, actingUserId, { card }); + return; + } + await thread.post({ card }); + } + // --------------------------------------------------------------------------- // Custom message handling (tool toMessage output) // --------------------------------------------------------------------------- diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts b/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts index 0313e12bd61..8bdc43eaee7 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts @@ -1,5 +1,5 @@ import type { AgentIntegrationConfig } from '@n8n/api-types'; -import type { ActionEvent, Thread } from 'chat'; +import type { ActionEvent, Author, Thread } from 'chat'; import type { Logger } from 'n8n-workflow'; import type { @@ -8,7 +8,7 @@ import type { PlatformAgentContext, SettleActionMessage, } from './agent-chat-integration'; -import { onceStatusHandle } from './agent-chat-integration'; +import { onceStatusHandle, postToUserOrThread } from './agent-chat-integration'; import type { AgentChatMessageContextBridge } from './agent-chat-message-context'; import type { AgentChatStreamConsumer } from './agent-chat-stream-consumer'; import type { CallbackStore } from './callback-store'; @@ -18,7 +18,14 @@ import type { AgentExecutionOrchestratorService, } from '../agent-execution-orchestrator.service'; -type ResumeExecutor = Pick; +type ResumeExecutor = Pick & { + /** Optional: a caller that cannot look checkpoints up simply skips the gate. */ + isResumable?(config: { agentId: string; runId: string }): Promise; +}; + +/** Covers both an expired callback key and a checkpoint that is gone. */ +const STALE_ACTION_NOTICE = + 'This action is no longer available. The link may have expired or already been used.'; interface AgentChatHitlResumeHandlerOptions { agentId: string; @@ -62,11 +69,29 @@ export class AgentChatHitlResumeHandler { return; } - const callbackData = await this.resolveCallbackData(event.actionId, event.value, thread); + const callbackData = await this.resolveCallbackData( + event.actionId, + event.value, + thread, + event.user, + ); if (!callbackData) return; const parsed = this.parseActionId(callbackData.actionId, callbackData.value); if (!parsed) return; + + // Resuming a gone run reports it as an agent misconfiguration, which it is + // not. Check before the card is settled, so a stale one is answered rather + // than relabelled with a decision that never took effect. + if (!(await this.isRunResumable(parsed.runId))) { + // Settling is not an option here: there is no decision to name. + if (this.options.deleteActionMessageBeforeResume) await this.deleteActionMessage(event); + await postToUserOrThread(thread, event.user, STALE_ACTION_NOTICE); + return; + } + // Persist the interacting user / messageId into the thread's message + // context so tools running on resume can read it via the message + // context store — no need to bolt a duplicate copy onto resumeData. const platformThreadId = this.options.resolvePlatformThreadId(thread); const threadId = this.options.toAgentThreadId(platformThreadId); const messageContext = this.options.messageContextBridge.capture(thread, { @@ -80,6 +105,7 @@ export class AgentChatHitlResumeHandler { await this.executeResume(thread, parsed.runId, parsed.toolCallId, parsed.resumeData, { messageContext, contextConversation: { threadId: threadId.id, resourceId: event.user.userId }, + actingUser: event.user, }); } @@ -127,10 +153,10 @@ export class AgentChatHitlResumeHandler { actionId: string, value: string | undefined, thread: Thread, + user: Author, ): Promise<{ actionId: string; value: string | undefined; - kind?: 'approval'; label?: string; } | null> { if (!this.options.callbackStore) return { actionId, value }; @@ -138,15 +164,12 @@ export class AgentChatHitlResumeHandler { const resolved = await this.options.callbackStore.resolve(actionId); if (!resolved) { this.options.logger.warn('[AgentChatBridge] Callback key not found or expired', { actionId }); - await thread.post( - 'This action is no longer available. The link may have expired or already been used.', - ); + await postToUserOrThread(thread, user, STALE_ACTION_NOTICE); return null; } return { actionId: resolved.actionId, value: resolved.value, - kind: resolved.kind, label: resolved.label, }; } @@ -155,22 +178,18 @@ export class AgentChatHitlResumeHandler { private async cleanUpBeforeResume( event: ActionEvent, resumeData: unknown, - callbackData: { kind?: 'approval'; label?: string }, + callbackData: { label?: string }, ): Promise { if (this.options.deleteActionMessageBeforeResume) { - try { - await event.adapter.deleteMessage(event.threadId, event.messageId); - } catch (deleteError) { - this.options.logger.warn('[AgentChatBridge] Failed to delete card message', { - error: deleteError instanceof Error ? deleteError.message : String(deleteError), - }); - } + await this.deleteActionMessage(event); return; } try { - const approved = - callbackData.kind === 'approval' ? this.getApprovalDecision(resumeData) : undefined; + // Read the decision off the payload, not off store metadata: the button + // value already carries `{ approved }` whenever the tool's resume schema + // declares it, so platforms without a CallbackStore get it too. + const approved = this.getApprovalDecision(resumeData); const message = this.options.formatActionDecisionMessage?.({ ...(approved !== undefined ? { approved } : {}), ...(callbackData.label !== undefined ? { selectedLabel: callbackData.label } : {}), @@ -197,6 +216,33 @@ export class AgentChatHitlResumeHandler { } } + private async deleteActionMessage(event: ActionEvent): Promise { + try { + await event.adapter.deleteMessage(event.threadId, event.messageId); + } catch (deleteError) { + this.options.logger.warn('[AgentChatBridge] Failed to delete card message', { + error: deleteError instanceof Error ? deleteError.message : String(deleteError), + }); + } + } + + private async isRunResumable(runId: string): Promise { + if (!this.options.agentService.isResumable) return true; + try { + return await this.options.agentService.isResumable({ + agentId: this.options.agentId, + runId, + }); + } catch (error) { + // A failed lookup must not swallow the click: let the resume decide. + this.options.logger.warn('[AgentChatBridge] Could not check whether a run is resumable', { + runId, + error: error instanceof Error ? error.message : String(error), + }); + return true; + } + } + private getApprovalDecision(resumeData: unknown): boolean | undefined { if ( typeof resumeData !== 'object' || @@ -223,13 +269,26 @@ export class AgentChatHitlResumeHandler { toolCallId: string, resumeData: unknown, options: Pick & { - notifyOnDuplicate?: boolean; + /** + * The user who clicked. The duplicate-click notice goes to them, and a + * card the resumed turn raises is addressed to them. + */ + actingUser?: Author; + /** + * Addressee for a card the resumed turn raises, where no user clicked. + * A server-driven resume names one so the card stays private, but it + * must not speak to them, so it never triggers the notice. + */ + cardRecipientId?: string; } = {}, ): Promise { - const { notifyOnDuplicate = true, ...context } = options; + const { actingUser, cardRecipientId, ...context } = options; + const cardUserId = actingUser?.userId ?? cardRecipientId; if (this.activeResumedRuns.has(runId)) { this.options.logger.warn('[AgentChatBridge] Run is already active', { runId, toolCallId }); - if (notifyOnDuplicate) await thread.post('This action has already been handled'); + if (actingUser) { + await postToUserOrThread(thread, actingUser, 'This action has already been handled'); + } return; } @@ -250,6 +309,7 @@ export class AgentChatHitlResumeHandler { await this.options.streamConsumer.consume(stream, thread, { ...resumeExecutionContext, statusHandle, + ...(cardUserId ? { actingUserId: cardUserId } : {}), }); } finally { // The stream consumer clears the status right before the first response; 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 46f792ddbfa..c9ec146f8d5 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-integration.ts @@ -104,6 +104,30 @@ export function onceStatusHandle( }; } +/** Narrower than `Thread.post`, which also takes an `AsyncIterable` to stream. */ +type EphemeralPostable = Parameters['postEphemeral']>[1]; + +/** + * Posts to `user` alone where the platform supports it, and to the thread where + * it does not, so the payload is never dropped. `fallbackToDM: false` because + * the SDK's own fallback would make this an unsolicited DM on Discord and + * Telegram. + */ +export async function postToUserOrThread( + thread: Thread, + user: string | Author, + payload: EphemeralPostable, +): Promise { + try { + const sent = await thread.postEphemeral(user, payload, { fallbackToDM: false }); + if (sent) return; + } catch { + // A rejected ephemeral post — rate limit, the user having left, a + // conversation that refuses targeting — must not cost the message. + } + await thread.post(payload); +} + export interface BridgeExecutionContext { platformAgentContext: PlatformAgentContext; slackThreadContext?: SlackThreadContext; @@ -251,6 +275,14 @@ export abstract class AgentChatIntegration { /** Whether action messages are deleted before the agent resumes. */ readonly deleteActionMessageBeforeResume: boolean = true; + /** + * True to deliver a suspension card only to the user whose turn raised it, + * so the rest of a channel never sees it. Delivery-scoped only: nothing + * verifies who clicks. The card still goes to the whole conversation where + * the platform has no ephemeral delivery, rather than being dropped. + */ + readonly targetSuspensionCardAtActingUser: boolean = false; + /** * True if the bridge should buffer streaming output and post it as a single * message instead of streaming text deltas via post-and-edit. 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 cb83938323b..2b42ac4f2c1 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 @@ -25,6 +25,7 @@ interface AgentChatStreamConsumerOptions { handleSuspension: ( chunk: SuspendedChunk, thread: Thread, + actingUserId?: string, ) => Promise; handleMessage: ( chunk: MessageChunk, @@ -44,6 +45,11 @@ interface ConsumeStreamOptions { /** Buffer output and report posting failures so the caller can retry. */ throwOnDeliveryError?: boolean; statusHandle?: BridgeStatusHandle; + /** + * The user whose message or click drove this turn, so a suspension card can + * be addressed to them. Absent for a turn no user drove. + */ + actingUserId?: string; } interface ResponseState { @@ -193,7 +199,7 @@ export class AgentChatStreamConsumer { } case 'tool-call-suspended': { await responseLifecycle.startDiscreteResponse(); - const result = await this.options.handleSuspension(chunk, thread); + const result = await this.options.handleSuspension(chunk, thread, options.actingUserId); responseState.hasVisibleResponse ||= result === 'posted'; if (result === 'failed') { responseState.fallbackSource = 'suspension'; @@ -361,7 +367,7 @@ export class AgentChatStreamConsumer { await flushBuffer(); } await responseLifecycle.startDiscreteResponse(); - const result = await this.options.handleSuspension(chunk, thread); + const result = await this.options.handleSuspension(chunk, thread, options.actingUserId); responseState.hasVisibleResponse ||= result === 'posted'; if (result === 'failed') { if (options.throwOnDeliveryError) { diff --git a/packages/cli/src/modules/agents/integrations/callback-store.ts b/packages/cli/src/modules/agents/integrations/callback-store.ts index 4ba6a06cc23..04d277e6e52 100644 --- a/packages/cli/src/modules/agents/integrations/callback-store.ts +++ b/packages/cli/src/modules/agents/integrations/callback-store.ts @@ -7,12 +7,11 @@ import type { CacheService } from '@/services/cache/cache.service'; export interface CallbackPayload { actionId: string; value: string; - kind?: 'approval'; groupId?: string; label?: string; } -export type CallbackMetadata = Pick; +export type CallbackMetadata = Pick; type CallbackStoreMetadata = CallbackMetadata & Pick; const DEFAULT_TTL_MS = 60 * 60 * 1000; 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 c44036715b6..ae2a60a615d 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 @@ -5,8 +5,10 @@ import { cardAction, dmFollowUp, dmMessage, + groupChatMention, selfMessage, TEAMS_DM_CONVERSATION_ID, + TEAMS_GROUP_CHAT_CONVERSATION_ID, TEAMS_USER_ID, } from '../../../__tests__/helpers/teams/synthetic-fixtures'; @@ -150,6 +152,14 @@ describe('Microsoft Teams integration scenarios', () => { displayName: 'Send Teams message', args: { text: 'Continue?' }, }, + // What the approval gate really sends: APPROVAL_RESUME_SCHEMA as + // JSON Schema. Without it the mapper encodes the button value as a + // bare string, and no decision reaches the formatter. + resumeSchema: { + type: 'object', + properties: { approved: { type: 'boolean' } }, + required: ['approved'], + }, }, { type: 'finish', finishReason: 'stop' }, ], @@ -175,13 +185,155 @@ describe('Microsoft Teams integration scenarios', () => { runId: 'run-card-1', toolCallId: 'tool-card-1', integrationType: 'teams', + resumeData: { approved: true }, }), ); - // Generic wording: no CallbackStore, so no decision reaches the formatter. - expect(ctx.lastEdit()?.body).toMatchObject({ - text: '✅ Action selected by Alice', + // Removed rather than relabelled, so a stale button cannot be clicked. + expect(ctx.lastDelete()?.body.uri).toContain(cardMessageId); + expect(ctx.lastEdit()).toBeUndefined(); + } finally { + await ctx.shutdown(); + } + }); + + it('addresses a group chat approval card at the user who asked', async () => { + const ctx = await createTeamsReplayContext({ + stream: [ + { + type: 'tool-call-suspended', + runId: 'run-group-1', + toolCallId: 'tool-group-1', + toolName: 'approval', + suspendPayload: { + type: 'approval', + toolName: 'send_teams_message', + displayName: 'Send Teams message', + args: { text: 'Continue?' }, + }, + resumeSchema: { + type: 'object', + properties: { approved: { type: 'boolean' } }, + required: ['approved'], + }, + }, + { type: 'finish', finishReason: 'stop' }, + ], + }); + try { + await ctx.sendWebhook(groupChatMention); + + const cardPost = ctx.lastPost(); + expect(cardPost?.body).toMatchObject({ + conversation: { id: TEAMS_GROUP_CHAT_CONVERSATION_ID }, + // A Teams targeted message: delivered to this user alone. + recipient: { id: TEAMS_USER_ID }, }); + expect(cardActions(cardPost?.body)).not.toHaveLength(0); + } finally { + await ctx.shutdown(); + } + }); + + it('removes an answered targeted card', async () => { + const ctx = await createTeamsReplayContext({ + stream: [ + { + type: 'tool-call-suspended', + runId: 'run-group-1', + toolCallId: 'tool-group-1', + toolName: 'approval', + suspendPayload: { + type: 'approval', + toolName: 'send_teams_message', + displayName: 'Send Teams message', + args: { text: 'Continue?' }, + }, + resumeSchema: { + type: 'object', + properties: { approved: { type: 'boolean' } }, + required: ['approved'], + }, + }, + { type: 'finish', finishReason: 'stop' }, + ], + }); + try { + await ctx.sendWebhook(groupChatMention); + const cardMessageId = ctx.lastPostedMessageId(); + if (!cardMessageId) throw new Error('Expected the approval card to have been posted'); + const approve = cardActions(ctx.lastPost()?.body)[0]; + if (!approve) throw new Error('Expected an Adaptive Card action on the approval card'); + + ctx.nextStream([{ type: 'finish', finishReason: 'stop' }]); + await ctx.sendWebhook( + cardAction(approve.data as Record, cardMessageId, groupChatMention), + ); + + expect(ctx.agentExecutor.resumeForChat).toHaveBeenCalledWith( + expect.objectContaining({ + runId: 'run-group-1', + toolCallId: 'tool-group-1', + resumeData: { approved: true }, + }), + ); + + // A targeted card is only mutable through the targeted endpoint, so the + // answered one is deleted there rather than settled in place. + expect(ctx.lastDelete()?.body.uri).toContain(cardMessageId); + expect(ctx.lastDelete()?.body.uri).toContain('isTargetedActivity=true'); + expect(ctx.lastEdit()).toBeUndefined(); + } finally { + await ctx.shutdown(); + } + }); + + it('answers a click on a card whose run is gone, privately', async () => { + const ctx = await createTeamsReplayContext({ + stream: [ + { + type: 'tool-call-suspended', + runId: 'run-group-1', + toolCallId: 'tool-group-1', + toolName: 'approval', + suspendPayload: { + type: 'approval', + toolName: 'send_teams_message', + displayName: 'Send Teams message', + args: { text: 'Continue?' }, + }, + resumeSchema: { + type: 'object', + properties: { approved: { type: 'boolean' } }, + required: ['approved'], + }, + }, + { type: 'finish', finishReason: 'stop' }, + ], + }); + try { + await ctx.sendWebhook(groupChatMention); + const cardMessageId = ctx.lastPostedMessageId(); + if (!cardMessageId) throw new Error('Expected the approval card to have been posted'); + const approve = cardActions(ctx.lastPost()?.body)[0]; + if (!approve) throw new Error('Expected an Adaptive Card action on the approval card'); + + ctx.agentExecutor.isResumable.mockResolvedValue(false); + await ctx.sendWebhook( + cardAction(approve.data as Record, cardMessageId, groupChatMention), + ); + + // No decision took effect, so the click is answered rather than resumed. + expect(ctx.agentExecutor.resumeForChat).not.toHaveBeenCalled(); + expect(ctx.lastDelete()?.body.uri).toContain(cardMessageId); + + const notice = ctx.lastPost(); + expect(notice?.body).toMatchObject({ + conversation: { id: TEAMS_GROUP_CHAT_CONVERSATION_ID }, + // Targeted, so the rest of the group chat never sees it. + recipient: { id: TEAMS_USER_ID }, + }); + expect(notice?.body.text).toContain('This action is no longer available'); } 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 7890a621cd2..b2652a823c8 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,7 +10,6 @@ import { AgentChatIntegration, type AgentChannelPreconditionContext, type AgentChatIntegrationContext, - type ActionDecisionMessageParams, } from '../../agent-chat-integration'; import { expandSelectsToButtons, type SuspendComponent } from '../../component-mapper'; import { assertCredentialNotClaimed } from '../../credential-claim'; @@ -88,10 +87,16 @@ export class TeamsIntegration extends AgentChatIntegration { ]; /** - * Teams acknowledges an Adaptive Card action by editing the card in place, so - * the answered card is settled rather than deleted. + * A channel or group chat card goes out as a Teams targeted message, so the + * rest of the channel never sees the approval. Delivery-scoped only: nothing + * verifies who clicks. + * + * Deleting the answered card relies on + * `patches/@chat-adapter__teams@4.37.0.patch`, because the adapter mutates a + * targeted activity without `?isTargetedActivity=true` and Teams answers + * 400. Drop the patch once upstream sends the flag (vercel/chat#950). */ - readonly deleteActionMessageBeforeResume = false; + readonly targetSuspensionCardAtActingUser = true; readonly disableStreaming = true; @@ -140,18 +145,6 @@ export class TeamsIntegration extends AgentChatIntegration { return expandSelectsToButtons(components); } - formatActionDecisionMessage({ - approved, - selectedLabel, - user, - }: ActionDecisionMessageParams): string { - const responder = user.fullName || user.userName || user.userId; - if (approved === undefined) { - return `✅ ${selectedLabel || 'Action'} selected by ${responder}`; - } - return approved ? `✅ Approved by ${responder}` : `🚫 Declined by ${responder}`; - } - /** * 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 diff --git a/patches/@chat-adapter__teams@4.37.0.patch b/patches/@chat-adapter__teams@4.37.0.patch new file mode 100644 index 00000000000..3427db72298 --- /dev/null +++ b/patches/@chat-adapter__teams@4.37.0.patch @@ -0,0 +1,130 @@ +diff --git a/dist/index.js b/dist/index.js +index 5278d648103068f1f3875c27e42c775ab02238ed..c0f0a7302a731e79951a972f6e18e91b90b066c9 100644 +--- a/dist/index.js ++++ b/dist/index.js +@@ -1622,6 +1622,9 @@ var MESSAGEID_STRIP_PATTERN2 = /;messageid=\d+/; + var CACHE_TTL_MS = 30 * 24 * 60 * 60 * 1e3; + var USER_INFO_CACHE_TTL_MS = 60 * 60 * 1e3; + var USER_INFO_NEGATIVE_CACHE_TTL_MS = 5 * 60 * 1e3; ++// n8n patch (upstream: vercel/chat#950) — Teams deletes a targeted message 24 ++// hours after it is sent, so a record of one is worthless past that point. ++var TARGETED_ACTIVITY_TTL_MS = 24 * 60 * 60 * 1e3; + var USER_INFO_NEGATIVE_SENTINEL = "unresolvable"; + var DEFAULT_DIALOG_OPEN_TIMEOUT_MS = 5e3; + var TEAMS_REACTION_ALIASES = { +@@ -2417,6 +2420,7 @@ var TeamsAdapter = class { + }); + try { + const sent = await this.app.send(conversationId, activity2); ++ await this.rememberTargetedActivity(conversationId, sent.id || ""); + return { + id: sent.id || "", + threadId, +@@ -2449,6 +2453,7 @@ var TeamsAdapter = class { + }); + try { + const sent = await this.app.send(conversationId, activity); ++ await this.rememberTargetedActivity(conversationId, sent.id || ""); + this.logger.debug("Teams API: targeted send response", { + messageId: sent.id + }); +@@ -2467,6 +2472,61 @@ var TeamsAdapter = class { + handleTeamsError(error, "postEphemeral"); + } + } ++ // n8n patch (upstream: vercel/chat#950) — Teams accepts an update or a delete ++ // for a targeted activity only through ?isTargetedActivity=true; the plain ++ // endpoint answers 400. Nothing on the wire says afterwards which activities ++ // were targeted and editMessage/deleteMessage receive only a message id, so ++ // the ids sent targeted are recorded here and read back on mutation. ++ // ++ // The record is only a fast path. n8n backs the Chat state adapter with ++ // process memory, so a restart or a second main loses it — mutateActivity ++ // therefore tries the other endpoint rather than trusting the lookup. ++ targetedActivityKey(conversationId, messageId) { ++ return `teams:targetedActivity:${conversationId}:${messageId}`; ++ } ++ async rememberTargetedActivity(conversationId, messageId) { ++ if (!(this.chat && messageId)) return; ++ try { ++ await this.chat.getState().set( ++ this.targetedActivityKey(conversationId, messageId), ++ "1", ++ TARGETED_ACTIVITY_TTL_MS ++ ); ++ } catch { ++ // A state adapter that cannot write must not fail the send. ++ } ++ } ++ async isTargetedActivity(conversationId, messageId) { ++ if (!(this.chat && messageId)) return false; ++ try { ++ return await this.chat.getState().get( ++ this.targetedActivityKey(conversationId, messageId) ++ ) === "1"; ++ } catch { ++ return false; ++ } ++ } ++ async forgetTargetedActivity(conversationId, messageId) { ++ if (!(this.chat && messageId)) return; ++ try { ++ await this.chat.getState().delete( ++ this.targetedActivityKey(conversationId, messageId) ++ ); ++ } catch { ++ // The TTL clears it anyway. ++ } ++ } ++ // A lost record must not leave an answered card in the channel, so fall back ++ // to the endpoint we did not pick. A 1:1 chat has no targeted variant, so its ++ // failure is real and propagates. ++ async mutateActivity(threadId, preferTargeted, targetedCall, plainCall) { ++ try { ++ return await (preferTargeted ? targetedCall() : plainCall()); ++ } catch (error) { ++ if (this.isDM(threadId)) throw error; ++ return await (preferTargeted ? plainCall() : targetedCall()); ++ } ++ } + createTargetedRecipient(userId) { + return { + id: userId, +@@ -2511,7 +2571,8 @@ var TeamsAdapter = class { + messageId + }); + try { +- await this.app.api.conversations.activities(conversationId).update(messageId, activity2); ++ const activities2 = this.app.api.conversations.activities(conversationId); ++ await this.mutateActivity(threadId, !this.isDM(threadId) && await this.isTargetedActivity(conversationId, messageId), () => activities2.updateTargeted(messageId, activity2), () => activities2.update(messageId, activity2)); + } catch (error) { + this.logger.error("Teams API: updateActivity failed", { + conversationId, +@@ -2534,7 +2595,8 @@ var TeamsAdapter = class { + textLength: text.length + }); + try { +- await this.app.api.conversations.activities(conversationId).update(messageId, activity); ++ const activities2 = this.app.api.conversations.activities(conversationId); ++ await this.mutateActivity(threadId, !this.isDM(threadId) && await this.isTargetedActivity(conversationId, messageId), () => activities2.updateTargeted(messageId, activity), () => activities2.update(messageId, activity)); + } catch (error) { + this.logger.error("Teams API: updateActivity failed", { + conversationId, +@@ -2553,7 +2615,8 @@ var TeamsAdapter = class { + messageId + }); + try { +- await this.app.api.conversations.activities(conversationId).delete(messageId); ++ const activities2 = this.app.api.conversations.activities(conversationId); ++ await this.mutateActivity(threadId, !this.isDM(threadId) && await this.isTargetedActivity(conversationId, messageId), () => activities2.deleteTargeted(messageId), () => activities2.delete(messageId)); + } catch (error) { + this.logger.error("Teams API: deleteActivity failed", { + conversationId, +@@ -2562,6 +2625,7 @@ var TeamsAdapter = class { + }); + handleTeamsError(error, "deleteMessage"); + } ++ if (!this.isDM(threadId)) await this.forgetTargetedActivity(conversationId, messageId); + this.logger.debug("Teams API: deleteActivity response", { ok: true }); + } + async addReaction(threadId, messageId, emoji) { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 11bed4ee167..31894c1a45d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1053,6 +1053,7 @@ overrides: strnum@<2.4.2: 2.4.2 patchedDependencies: + '@chat-adapter/teams@4.37.0': af5fb1d38a2af55399bd8b6934e77348a9b061fec6177c101dd9d7a787ae8a6c '@confluentinc/kafka-javascript@1.9.1': c597d3e3d9d6d0d33588b7dc1e88d55f270ba155721aad464bd7fb03e28f5003 '@langchain/google-common@2.1.24': b50436a6c7b0a82d54472dff331ba1b42ab8b816995f83295eef2ee4b482a87d '@langchain/openai@1.4.4': e16bcc93cf9037198faacf494fb81afba6e9abba749dc9a09c3abfb33e68d5c2 @@ -4407,7 +4408,7 @@ importers: version: 4.37.0(ai@7.0.74(zod@3.25.76))(supports-color@8.1.1)(zod@3.25.76) '@chat-adapter/teams': specifier: 'catalog:' - version: 4.37.0(ai@7.0.74(zod@3.25.76))(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)(zod@3.25.76) + version: 4.37.0(patch_hash=af5fb1d38a2af55399bd8b6934e77348a9b061fec6177c101dd9d7a787ae8a6c)(ai@7.0.74(zod@3.25.76))(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)(zod@3.25.76) '@chat-adapter/telegram': specifier: 'catalog:' version: 4.37.0(ai@7.0.74(zod@3.25.76))(supports-color@8.1.1)(zod@3.25.76) @@ -27917,7 +27918,7 @@ snapshots: - workflow - zod - '@chat-adapter/teams@4.37.0(ai@7.0.74(zod@3.25.76))(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)(zod@3.25.76)': + '@chat-adapter/teams@4.37.0(patch_hash=af5fb1d38a2af55399bd8b6934e77348a9b061fec6177c101dd9d7a787ae8a6c)(ai@7.0.74(zod@3.25.76))(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)(zod@3.25.76)': dependencies: '@chat-adapter/shared': 4.37.0(ai@7.0.74(zod@3.25.76))(supports-color@8.1.1)(zod@3.25.76) '@microsoft/teams.api': 2.0.16(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 6f662c515ce..9673fdac1d3 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -500,6 +500,7 @@ patchedDependencies: '@langchain/google-common@2.1.24': patches/@langchain__google-common@2.1.24.patch '@confluentinc/kafka-javascript@1.9.1': patches/@confluentinc__kafka-javascript@1.9.1.patch imapflow@1.7.1: patches/imapflow@1.7.1.patch + '@chat-adapter/teams@4.37.0': patches/@chat-adapter__teams@4.37.0.patch allowBuilds: '@biomejs/biome': false