fix(core): recover missed queued resume updates

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Robin Braumann 2026-09-15 09:17:02 +02:00
parent da6019181b
commit bbcd70bad8
No known key found for this signature in database
3 changed files with 98 additions and 9 deletions

View File

@ -11,6 +11,7 @@ import { UserError, type Logger } from 'n8n-workflow';
import { CacheService } from '@/services/cache/cache.service';
import { AgentTurnQueueService, type AgentTurnClaim } from '../../agent-turn-queue.service';
import type { AgentExecution } from '../../entities/agent-execution.entity';
import type { AgentRepository } from '../../repositories/agent.repository';
import { AgentChatBridge } from '../agent-chat-bridge';
import {
@ -64,6 +65,8 @@ function makeBot() {
handlers.slashCommand = h;
},
getAdapter: vi.fn().mockReturnValue(undefined),
getState: vi.fn().mockReturnValue({}),
getLogger: vi.fn().mockReturnValue(undefined),
thread: vi.fn(),
};
return { bot, handlers };
@ -2629,6 +2632,85 @@ describe('AgentChatBridge — consumeStream', () => {
expect(thread.post).toHaveBeenCalledWith({ markdown: 'Approved response' });
expect(agentExecutor.resumeForChat).toHaveBeenCalled();
});
it('reconstructs and settles a queued channel action before resuming', async () => {
const chatSdk = await import('chat');
const loadChatSdkSpy = vi.spyOn(esmLoader, 'loadChatSdk').mockResolvedValue(chatSdk);
const { bot } = makeBot();
const deleteMessage = vi.fn().mockResolvedValue(undefined);
bot.getAdapter.mockReturnValue({ deleteMessage });
const agentExecutor = makeAgentExecutor([finishChunk]);
const messageContextStore = mock<IntegrationMessageContextService>();
const bridge = new AgentChatBridge(
bot as unknown as ChatBotLike,
'agent-1',
agentExecutor as never,
componentMapper,
logger,
'project-1',
streamingIntegration,
messageContextStore,
);
const actionMessage = new chatSdk.Message({
id: 'card-message-1',
threadId: 'thread-1',
text: 'Approve?',
formatted: chatSdk.parseMarkdown('Approve?'),
raw: {},
author: {
userId: 'u2',
userName: 'user2',
fullName: 'User Two',
isBot: false,
isMe: false,
},
metadata: { dateSent: new Date('2026-01-01T00:00:00.000Z'), edited: false },
attachments: [],
});
const claim = claimFor('agent-1:thread-1');
const row = {
id: 'execution-queued',
runContext: {
kind: 'resume',
runId: 'run-1',
toolCallId: 'tool-1',
resumeData: { approved: true },
channel: {
integrationType: 'test-streaming',
credentialId: 'cred-1',
thread: {
_type: 'chat:Thread',
adapterName: 'test-streaming',
channelId: 'channel-1',
currentMessage: actionMessage.toJSON(),
id: 'thread-1',
isDM: false,
},
action: { actionId: 'resume:run-1:tool-1:0' },
},
},
} as AgentExecution;
try {
await bridge.runQueuedResume(row, claim);
} finally {
loadChatSdkSpy.mockRestore();
}
expect(deleteMessage).toHaveBeenCalledWith('thread-1', 'card-message-1');
expect(messageContextStore.setLatest).toHaveBeenCalledWith(
'agent-1:thread-1',
'u2',
expect.objectContaining({
messageId: 'card-message-1',
interactingUserId: 'u2',
}),
);
expect(agentExecutor.resumeForChat).toHaveBeenCalledWith(
expect.objectContaining({ runId: 'run-1', toolCallId: 'tool-1' }),
claim,
);
});
});
describe('message context', () => {

View File

@ -356,7 +356,17 @@ describe('useAgentChatStream — SDK-aligned event handling', () => {
expect(getTestChatMessagesMock).not.toHaveBeenCalled();
getTestChatMessagesMock.mockResolvedValue({ messages: [], openSuspensions: [] });
hook.refresh();
for (const listener of [...pushListeners]) {
listener({
type: 'agentExecutionUpdated',
data: {
projectId: 'p1',
agentId: 'a1',
threadId: 'thread-1',
executionId: 'exec-queued-resume',
},
});
}
await flushPromises();
expect(getTestChatMessagesMock).toHaveBeenCalledOnce();
});

View File

@ -107,7 +107,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
const pendingQueuedResumes = new Set<string>();
const executionTransitionsSeenDuringStream = new Map<
string,
NonNullable<AgentPersistedMessageDto['executionStatus']>
AgentPersistedMessageDto['executionStatus']
>();
let retryCount = 0;
let retryTimer: ReturnType<typeof setTimeout> | undefined;
@ -222,14 +222,10 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
let pendingResumeSettled = false;
if (event) {
const { executionId, executionStatus } = event.data;
if (
pendingQueuedResumes.has(executionId) &&
executionStatus !== undefined &&
executionStatus !== 'queued'
) {
if (pendingQueuedResumes.has(executionId) && executionStatus !== 'queued') {
pendingQueuedResumes.delete(executionId);
pendingResumeSettled = true;
} else if (isStreaming.value && executionStatus !== undefined) {
} else if (isStreaming.value) {
executionTransitionsSeenDuringStream.set(executionId, executionStatus);
}
}
@ -755,9 +751,10 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
session.submittedUserMessage.executionId = event.executionId;
}
if (session.submittedResumeToolCallId) {
const transitionSeen = executionTransitionsSeenDuringStream.has(event.executionId);
const latestStatus = executionTransitionsSeenDuringStream.get(event.executionId);
executionTransitionsSeenDuringStream.delete(event.executionId);
if (latestStatus === undefined || latestStatus === 'queued') {
if (!transitionSeen || latestStatus === 'queued') {
pendingQueuedResumes.add(event.executionId);
}
}