fix(core): Make a Microsoft Teams approval card private and self-clearing (no-changelog) (#39515)

This commit is contained in:
Bernhard Wittmann 2026-09-25 10:13:14 +00:00 committed by GitHub
parent c88df6f9c7
commit e64291dc6d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 987 additions and 160 deletions

View File

@ -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<SerializableAgentState>({ status });
}
describe('AgentConversationStateService', () => {
describe('isResumable', () => {
const executionRepository = mock<AgentExecutionRepository>();
function createService(status: Awaited<ReturnType<N8NCheckpointStorage['getStatus']>>) {
const checkpointStorage = mock<N8NCheckpointStorage>({
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);
});
});
});

View File

@ -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);

View File

@ -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<boolean> {
const result = await this.checkpointStorage.getStatus(runId, agentId);
return result.status === 'active' && result.checkpoint.status === 'suspended';
}
}

View File

@ -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<StreamChunk>,
abortSignal: AbortSignal,
delivery?: { bridge: AgentChatBridge; threadId: string },
cardRecipientId?: string,
): AsyncGenerator<StreamChunk> {
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(

View File

@ -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<typeof AgentChatHitlResumeHandler>[0];
function createHandler(overrides: Partial<HandlerOptions> = {}) {
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<void>((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<void>((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 } }),
);
});
});

View File

@ -27,6 +27,13 @@ function makeConsumer(
const thread = mock<Thread<unknown, unknown>>();
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<Logger>(),
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);
});
});

View File

@ -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"]',
});
});

View File

@ -99,6 +99,7 @@ export interface LinearReplayContext extends Omit<ReplayContextSetup, 'nextStrea
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
isResumable: Mock;
};
apiCalls: LinearApiCall[];
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];

View File

@ -189,6 +189,7 @@ export interface ReplayContextSetup<TChat extends ChatInstance = ChatInstance> {
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
isResumable: Mock;
};
actionExecutor: ChatIntegrationActionExecutor;
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
@ -231,6 +232,9 @@ export function createReplayContextSetup<TChat extends ChatInstance>(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 }> = [];

View File

@ -42,6 +42,7 @@ export interface TeamsReplayContext extends Omit<ReplayContextSetup, 'chat'> {
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();

View File

@ -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<string, unknown>, replyToId: string) {
export function cardAction(
value: Record<string, unknown>,
replyToId: string,
from: TeamsActivityFixture = baseActivity(),
) {
return baseActivity({
id: 'activity-action-1',
conversation: from.conversation,
channelData: from.channelData,
replyToId,
value,
});
}
const mentionEntity = {
type: 'mention',
text: '<at>n8n Agent</at>',
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> = {}): 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: '<at>n8n Agent</at> hello agent',
entities: [mentionEntity],
});

View File

@ -92,6 +92,7 @@ export interface TelegramReplayContext extends Omit<ReplayContextSetup, 'nextStr
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
isResumable: Mock;
};
actionExecutor: ChatIntegrationActionExecutor;
apiCalls: TelegramApiCall[];

View File

@ -63,7 +63,7 @@ describe('TeamsIntegration', () => {
});
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<typeof integration.formatActionDecisionMessage>[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<typeof integration.formatActionDecisionMessage>[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<typeof integration.formatActionDecisionMessage>[0]),
).toBe('✅ Action selected by Alice');
});
});
describe('normalizeComponents', () => {
it('converts select options into individual buttons', () => {
expect(

View File

@ -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<AgentExecutionOrchestratorService, 'resumeF
agentId: string;
threadId: string;
}): Promise<OpenSuspension | null>;
/** Optional: a caller that cannot look checkpoints up simply skips the gate. */
isResumable?(config: { agentId: string; runId: string }): Promise<boolean>;
}
/** 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<void> {
async deliverWakeResponse(
threadId: string,
chunks: StreamChunk[],
cardRecipientId?: string,
): Promise<void> {
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<StreamChunk, { type: 'tool-call-suspended' }>,
thread: Thread,
actingUserId?: string,
): Promise<SuspensionHandlingResult> {
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<void> {
if (this.integrationImpl?.targetSuspensionCardAtActingUser && actingUserId) {
await postToUserOrThread(thread, actingUserId, { card });
return;
}
await thread.post({ card });
}
// ---------------------------------------------------------------------------
// Custom message handling (tool toMessage output)
// ---------------------------------------------------------------------------

View File

@ -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<AgentExecutionOrchestratorService, 'resumeForChat'>;
type ResumeExecutor = Pick<AgentExecutionOrchestratorService, 'resumeForChat'> & {
/** Optional: a caller that cannot look checkpoints up simply skips the gate. */
isResumable?(config: { agentId: string; runId: string }): Promise<boolean>;
};
/** 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<unknown, unknown>,
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<void> {
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<void> {
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<boolean> {
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<ResumeForChatConfig, 'messageContext' | 'contextConversation'> & {
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<void> {
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;

View File

@ -104,6 +104,30 @@ export function onceStatusHandle(
};
}
/** Narrower than `Thread.post`, which also takes an `AsyncIterable` to stream. */
type EphemeralPostable = Parameters<Thread<unknown, unknown>['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<unknown, unknown>,
user: string | Author,
payload: EphemeralPostable,
): Promise<void> {
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.

View File

@ -25,6 +25,7 @@ interface AgentChatStreamConsumerOptions {
handleSuspension: (
chunk: SuspendedChunk,
thread: Thread<unknown, unknown>,
actingUserId?: string,
) => Promise<SuspensionHandlingResult>;
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) {

View File

@ -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<CallbackPayload, 'kind' | 'groupId'>;
export type CallbackMetadata = Pick<CallbackPayload, 'groupId'>;
type CallbackStoreMetadata = CallbackMetadata & Pick<CallbackPayload, 'label'>;
const DEFAULT_TTL_MS = 60 * 60 * 1000;

View File

@ -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<string, unknown>, 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<string, unknown>, 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();
}

View File

@ -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

View File

@ -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) {

View File

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

View File

@ -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