mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
feat(editor): Add queued message controls to agent Preview chat (#39487)
This commit is contained in:
Robin Braumann
GitHub
parent
6437496faf
commit
8a4fd13423
|
|
@ -73,7 +73,8 @@ export type ForwardedChildChunkWire =
|
|||
};
|
||||
|
||||
export type AgentSseEvent =
|
||||
| { type: 'execution-started'; executionId: string; sessionId: string }
|
||||
| { type: 'message-queued'; queueId: string; sessionId: string }
|
||||
| { type: 'execution-started'; executionId: string; sessionId: string; message?: string }
|
||||
| { type: 'start-step' }
|
||||
| { type: 'finish-step' }
|
||||
| { type: 'text-start'; id: string }
|
||||
|
|
|
|||
|
|
@ -260,6 +260,10 @@ export class AgentChatMessageDto extends Z.class(agentChatMessageShape) {
|
|||
}
|
||||
}
|
||||
|
||||
export class AgentChatQueueUpdateDto extends Z.class({
|
||||
message: z.string(),
|
||||
}) {}
|
||||
|
||||
export class AgentChatResumeDto extends Z.class({
|
||||
runId: z.string().min(1),
|
||||
toolCallId: z.string().min(1),
|
||||
|
|
|
|||
|
|
@ -380,6 +380,17 @@ export interface AgentBuilderOpenSuspension {
|
|||
suspendPayload?: unknown;
|
||||
}
|
||||
|
||||
export interface AgentChatQueueItem {
|
||||
id: string;
|
||||
message: string;
|
||||
attachments?: Array<{ id: string; fileName: string; mimeType: string; sizeBytes: number }>;
|
||||
createdAt: string;
|
||||
}
|
||||
|
||||
export interface AgentChatQueueResponse {
|
||||
items: AgentChatQueueItem[];
|
||||
}
|
||||
|
||||
/** Chat history envelope returned by the agent chat messages endpoints. */
|
||||
export interface AgentChatMessagesResponse {
|
||||
messages: AgentPersistedMessageDto[];
|
||||
|
|
|
|||
|
|
@ -29,4 +29,13 @@ export type AgentUpdated = {
|
|||
};
|
||||
};
|
||||
|
||||
export type AgentsPushMessage = AgentExecutionUpdated | AgentUpdated | AgentBackgroundJobsUpdated;
|
||||
export type AgentMessageQueueUpdated = {
|
||||
type: 'agentMessageQueueUpdated';
|
||||
data: { projectId: string; agentId: string; threadId: string };
|
||||
};
|
||||
|
||||
export type AgentsPushMessage =
|
||||
| AgentExecutionUpdated
|
||||
| AgentUpdated
|
||||
| AgentBackgroundJobsUpdated
|
||||
| AgentMessageQueueUpdated;
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ export type PubSubEventName =
|
|||
| 'relay-agent-execution-update'
|
||||
| 'relay-agent-queued-chat'
|
||||
| 'cancel-agent-chat-execution'
|
||||
| 'relay-agent-message-queue-update'
|
||||
| 'relay-agent-background-tasks-update'
|
||||
| 'relay-agent-update'
|
||||
| 'resume-agent-workflow-tool'
|
||||
|
|
|
|||
|
|
@ -1,15 +1,19 @@
|
|||
# Preview chat recovery
|
||||
|
||||
An accepted preview turn continues when its browser connection closes. This applies to new messages and resumed interactions. The server records the execution and registers its cancellation handle before it sends `execution-started`. The event contains `executionId` and `sessionId`.
|
||||
An accepted preview turn continues when its browser connection closes. This applies to new messages and resumed interactions. For a new message, the server commits the session and queue item before it sends `message-queued`. This event contains `queueId` and `sessionId`. The composer clears the submitted draft on acceptance. When the consumer claims the item, it records the execution and registers its cancellation handle before it sends `execution-started`. The event contains `executionId` and `sessionId`.
|
||||
|
||||
The original connection receives live SSE events. After a reload or disconnect, the client reads the recorded transcript. It follows later `agentExecutionUpdated` notifications with fresh history reads. Each successful read replaces the recovered transcript. A failed read keeps the last known state. The client does not submit the message or resume request again.
|
||||
|
||||
`GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/messages` returns `messages`, `openSuspensions`, and `activeExecutionId`. The active ID is present even before the first output. A `null` ID means that the recorded thread has no running execution.
|
||||
|
||||
All tabs of the owning user can follow the execution and request Stop. Other users cannot read or cancel a private preview turn. New-message submission stays blocked while a turn runs. The draft remains editable. A competing request receives the `turn_already_running` error and keeps its draft.
|
||||
All tabs of the owning user can follow the execution and request Stop. Other users cannot read or cancel a private preview turn. Users can submit messages while a turn runs or waits for approval. Pending messages appear only in the queue panel. Each message enters the conversation when its execution starts. Typed answers to question cards still resume the suspended interaction. Waiting SSE requests do not block history recovery or approval actions.
|
||||
|
||||
Stop uses `DELETE /projects/:projectId/agents/v2/:agentId/chat/:threadId/executions/:executionId`. It requires `agent:execute` and the preview owner. The server cancels the local execution or relays the request to the main that owns it. The response `{ cancelRequested }` reports the request. Recorded state determines completion. A late request cannot cancel a later execution. Suspended turns use the existing checkpoint cancellation path.
|
||||
Stop uses `DELETE /projects/:projectId/agents/v2/:agentId/chat/:threadId/executions/:executionId`. It requires `agent:execute` and the preview owner. The server cancels the local execution or relays the request to the main that owns it. The response `{ cancelRequested }` reports the request. Recorded state determines completion. A late request cannot cancel a later execution. Suspended turns use the existing checkpoint cancellation path. Stop leaves pending messages in place. The next message starts after cancellation settles.
|
||||
|
||||
Recovery guarantees the latest retained timeline, followed by newer snapshots. Updates can arrive in batches. Snapshots do not contain original text-stream IDs or every transient live event. Child text and reasoning share the existing 4,000-character limit for each delegation. Text beyond that limit is not part of recovery.
|
||||
|
||||
This contract does not include token replay, queued messages, execution fencing, or survival of backend process failure. The existing Agents enablement controls the feature.
|
||||
`GET /projects/:projectId/agents/v2/:agentId/chat/:threadId/queue` restores pending messages in FIFO order. It requires `agent:read` and the Preview session owner. `DELETE` on the same path with `/:queueId` requires `agent:execute`. Removal locks the session and rejects an item that has started. Successful removal discards its attachments. Queue changes use `agentMessageQueueUpdated` notifications across mains. Execution updates also refresh the queue. Neither notification replays stream events.
|
||||
|
||||
Accepted input survives backend process failure through the durable queue. Interrupted executions are not replayed. Valid approval checkpoints retain their place. The existing Agents enablement controls the feature. This contract does not include token replay, sender deduplication, or queue reordering.
|
||||
|
||||
Users can edit the text of a pending message. Edits retain its attachments and queue position. The update uses the session lock and fails if the consumer has claimed the item. Queue notifications refresh the edited text across mains. The execution-start event carries the accepted text so waiting streams do not show the original draft. If a message starts during editing, the editor keeps the unsaved text available to copy until the user dismisses it.
|
||||
|
|
|
|||
|
|
@ -158,6 +158,9 @@ describe('AgentChatController route access scopes', () => {
|
|||
['cancelChatRun', 'agent:execute'],
|
||||
['cancelChatExecution', 'agent:execute'],
|
||||
['getChatMessages', 'agent:read'],
|
||||
['getQueuedMessages', 'agent:read'],
|
||||
['removeQueuedMessage', 'agent:execute'],
|
||||
['updateQueuedMessage', 'agent:execute'],
|
||||
['getBackgroundJobs', 'agent:read'],
|
||||
['getTestChatMessages', 'agent:read'],
|
||||
['clearTestChatMessages', 'agent:update'],
|
||||
|
|
@ -166,6 +169,31 @@ describe('AgentChatController route access scopes', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('AgentChatController queue editing', () => {
|
||||
it('reads the message after the request and response arguments supplied by the registry', async () => {
|
||||
const { controller, agentsService, messageQueue } = makeController();
|
||||
agentsService.findById.mockResolvedValue({ id: 'agent-1' } as never);
|
||||
const params = {
|
||||
projectId: 'project-1',
|
||||
agentId: 'agent-1',
|
||||
threadId: 'thread-1',
|
||||
queueId: '1',
|
||||
};
|
||||
|
||||
await Reflect.apply(controller.updateQueuedMessage, controller, [
|
||||
{ params, user: { id: 'user-1' } },
|
||||
makeSseResponse([]),
|
||||
{ message: 'Edited message' },
|
||||
]);
|
||||
|
||||
expect(messageQueue.updatePending).toHaveBeenCalledWith({
|
||||
...params,
|
||||
userId: 'user-1',
|
||||
message: 'Edited message',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('AgentChatController background tasks', () => {
|
||||
const thread = mock<AgentExecutionThread>({
|
||||
id: 'thread-1',
|
||||
|
|
@ -737,8 +765,14 @@ describe('AgentChatController attachment cleanup on failed turns', () => {
|
|||
expect(agentChatAttachmentService.deleteByIds).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('relays existing SSE events from the main that claims the message', async () => {
|
||||
it('acknowledges committed acceptance and relays events from the claiming main', async () => {
|
||||
const { controller, messageQueue, previewStreams } = makeController();
|
||||
const commit = createDeferredPromise();
|
||||
messageQueue.enqueue.mockImplementationOnce(async (_input, onInserted) => {
|
||||
onInserted?.('queue-1');
|
||||
await commit.promise;
|
||||
return mock<AgentMessageQueue>({ id: 'queue-1' });
|
||||
});
|
||||
const { res, events } = makeCleanupSseResponse();
|
||||
const request = controller.chat(
|
||||
{ params: { projectId: 'project-1' }, user: { id: 'user-1' } } as never,
|
||||
|
|
@ -747,6 +781,15 @@ describe('AgentChatController attachment cleanup on failed turns', () => {
|
|||
{ message: 'hi', sessionId: 'thread-1', newSession: true } as never,
|
||||
);
|
||||
await vi.waitFor(() => expect(messageQueue.enqueue).toHaveBeenCalled());
|
||||
expect(events()).toEqual([]);
|
||||
commit.resolve();
|
||||
await vi.waitFor(() =>
|
||||
expect(events()).toContainEqual({
|
||||
type: 'message-queued',
|
||||
queueId: 'queue-1',
|
||||
sessionId: 'thread-1',
|
||||
}),
|
||||
);
|
||||
const started = {
|
||||
type: 'execution-started' as const,
|
||||
executionId: 'exec-1',
|
||||
|
|
@ -758,7 +801,11 @@ describe('AgentChatController attachment cleanup on failed turns', () => {
|
|||
previewStreams.handleRelay({ queueId: 'queue-1', sequence: 2, event: done });
|
||||
previewStreams.handleRelay({ queueId: 'queue-1', sequence: 3, event: null });
|
||||
await request;
|
||||
expect(events()).toEqual([started, done]);
|
||||
expect(events()).toEqual([
|
||||
{ type: 'message-queued', queueId: 'queue-1', sessionId: 'thread-1' },
|
||||
started,
|
||||
done,
|
||||
]);
|
||||
expect(res.end).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -81,7 +81,7 @@ describe('AgentExecutionUpdateBroadcaster', () => {
|
|||
expect(publisher.publishCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('sends private execution and task updates only to the owner through the relay', async () => {
|
||||
it('sends private execution, task, and queue updates only to the owner through the relay', async () => {
|
||||
Object.defineProperty(instanceSettings, 'isWorker', { value: true, configurable: true });
|
||||
threadRepository.findOneBy.mockResolvedValue({
|
||||
...thread,
|
||||
|
|
@ -90,7 +90,8 @@ describe('AgentExecutionUpdateBroadcaster', () => {
|
|||
});
|
||||
broadcaster.notify(update);
|
||||
broadcaster.notifyBackgroundJobsUpdated('agent-1', 'thread-1');
|
||||
await vi.waitFor(() => expect(publisher.publishCommand).toHaveBeenCalledTimes(2));
|
||||
broadcaster.notifyQueueUpdated('thread-1');
|
||||
await vi.waitFor(() => expect(publisher.publishCommand).toHaveBeenCalledTimes(3));
|
||||
for (const [command] of publisher.publishCommand.mock.calls) {
|
||||
expect(command).toMatchObject({ payload: { userIds: ['user-2'] } });
|
||||
}
|
||||
|
|
@ -134,12 +135,13 @@ describe('AgentExecutionUpdateBroadcaster', () => {
|
|||
expect(publisher.publishCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('delivers relayed task updates without another relay', () => {
|
||||
it.each([
|
||||
['handleBackgroundJobsRelay', 'agentBackgroundTasksUpdated'],
|
||||
['handleQueueRelay', 'agentMessageQueueUpdated'],
|
||||
] as const)('delivers %s updates without another relay', (handler, type) => {
|
||||
const data = { projectId: 'project-1', agentId: 'agent-1', threadId: 'thread-1' };
|
||||
broadcaster.handleBackgroundJobsRelay({ data, userIds: ['user-2'] });
|
||||
expect(push.sendToUsers).toHaveBeenCalledWith({ type: 'agentBackgroundTasksUpdated', data }, [
|
||||
'user-2',
|
||||
]);
|
||||
broadcaster[handler]({ data, userIds: ['user-2'] });
|
||||
expect(push.sendToUsers).toHaveBeenCalledWith({ type, data }, ['user-2']);
|
||||
expect(publisher.publishCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -98,6 +98,8 @@ describe('AgentMessageQueueConsumer', () => {
|
|||
|
||||
it('runs independent sessions without waiting and reuses each claimed execution', async () => {
|
||||
const first = claim('first');
|
||||
first.item.payload.message = 'edited first message';
|
||||
first.recording.userMessage = first.item.payload.message;
|
||||
const second = claim('second');
|
||||
const waiting = createDeferredPromise();
|
||||
repository.findThreadIds.mockResolvedValue(['first', 'second']);
|
||||
|
|
@ -119,6 +121,12 @@ describe('AgentMessageQueueConsumer', () => {
|
|||
expect(queue.settle).toHaveBeenCalledWith('second', second.admission.executionId),
|
||||
);
|
||||
expect(queue.settle).not.toHaveBeenCalledWith('first', first.admission.executionId);
|
||||
expect(sender.send).toHaveBeenCalledWith({
|
||||
type: 'execution-started',
|
||||
executionId: first.admission.executionId,
|
||||
sessionId: 'first',
|
||||
message: 'edited first message',
|
||||
});
|
||||
expect(testRuns.prepareDraftRun).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
newSession: false,
|
||||
|
|
|
|||
|
|
@ -2,14 +2,25 @@ import {
|
|||
type AgentBackgroundJobsResponse,
|
||||
type AgentChatAttachmentPayload,
|
||||
AgentChatMessageDto,
|
||||
AgentChatQueueUpdateDto,
|
||||
type AgentChatMessagesResponse,
|
||||
type AgentChatQueueResponse,
|
||||
AgentChatResumeDto,
|
||||
MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES,
|
||||
MAX_AGENT_CHAT_ATTACHMENT_SIZE_MB,
|
||||
ViewableMimeTypes,
|
||||
} from '@n8n/api-types';
|
||||
import type { AuthenticatedRequest } from '@n8n/db';
|
||||
import { Body, Delete, Get, Param, Post, ProjectScope, RestController } from '@n8n/decorators';
|
||||
import {
|
||||
Body,
|
||||
Delete,
|
||||
Get,
|
||||
Param,
|
||||
Patch,
|
||||
Post,
|
||||
ProjectScope,
|
||||
RestController,
|
||||
} from '@n8n/decorators';
|
||||
import { scrubSecretsInText } from '@n8n/utils/scrub-secrets';
|
||||
import { sanitizeFilename } from '@n8n/utils/files/sanitize-filename';
|
||||
import type { Response } from 'express';
|
||||
|
|
@ -193,7 +204,7 @@ export class AgentChatController {
|
|||
});
|
||||
abortSignal.throwIfAborted();
|
||||
|
||||
await this.messageQueue.enqueue(
|
||||
const item = await this.messageQueue.enqueue(
|
||||
{
|
||||
agentId,
|
||||
projectId,
|
||||
|
|
@ -214,6 +225,7 @@ export class AgentChatController {
|
|||
);
|
||||
accepted = true;
|
||||
subscription?.accepted();
|
||||
send({ type: 'message-queued', queueId: item.id, sessionId: threadId });
|
||||
await subscription?.done;
|
||||
} catch (error) {
|
||||
// Committed messages own their attachments, including after a disconnect.
|
||||
|
|
@ -322,6 +334,55 @@ export class AgentChatController {
|
|||
return { cancelled };
|
||||
}
|
||||
|
||||
@Get('/:agentId/chat/:threadId/queue')
|
||||
@ProjectScope('agent:read')
|
||||
async getQueuedMessages(
|
||||
req: AuthenticatedRequest<{ projectId: string; agentId: string; threadId: string }>,
|
||||
): Promise<AgentChatQueueResponse> {
|
||||
const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId);
|
||||
if (!agent) throw new NotFoundError('Agent not found');
|
||||
return await this.messageQueue.listPending({ ...req.params, userId: req.user.id });
|
||||
}
|
||||
|
||||
@Patch('/:agentId/chat/:threadId/queue/:queueId')
|
||||
@ProjectScope('agent:execute')
|
||||
async updateQueuedMessage(
|
||||
req: AuthenticatedRequest<{
|
||||
projectId: string;
|
||||
agentId: string;
|
||||
threadId: string;
|
||||
queueId: string;
|
||||
}>,
|
||||
_res: Response,
|
||||
@Body payload: AgentChatQueueUpdateDto,
|
||||
): Promise<void> {
|
||||
if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID');
|
||||
const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId);
|
||||
if (!agent) throw new NotFoundError('Agent not found');
|
||||
await this.messageQueue.updatePending({
|
||||
...req.params,
|
||||
userId: req.user.id,
|
||||
message: payload.message,
|
||||
});
|
||||
}
|
||||
|
||||
@Delete('/:agentId/chat/:threadId/queue/:queueId')
|
||||
@ProjectScope('agent:execute')
|
||||
async removeQueuedMessage(
|
||||
req: AuthenticatedRequest<{
|
||||
projectId: string;
|
||||
agentId: string;
|
||||
threadId: string;
|
||||
queueId: string;
|
||||
}>,
|
||||
) {
|
||||
if (!/^[1-9]\d*$/.test(req.params.queueId)) throw new BadRequestError('Invalid queue ID');
|
||||
const agent = await this.agentsService.findById(req.params.agentId, req.params.projectId);
|
||||
if (!agent) throw new NotFoundError('Agent not found');
|
||||
await this.messageQueue.removePending({ ...req.params, userId: req.user.id });
|
||||
return { removed: true };
|
||||
}
|
||||
|
||||
@Get('/:agentId/chat/:threadId/background-tasks')
|
||||
@ProjectScope('agent:read')
|
||||
async getBackgroundJobs(
|
||||
|
|
|
|||
|
|
@ -77,6 +77,32 @@ export class AgentExecutionUpdateBroadcaster {
|
|||
}
|
||||
}
|
||||
|
||||
notifyQueueUpdated(threadId: string): void {
|
||||
void this.broadcastQueueUpdated(threadId).catch((error: unknown) => {
|
||||
this.logger.warn('Failed to broadcast agent queue update', { threadId, error });
|
||||
});
|
||||
}
|
||||
|
||||
private async broadcastQueueUpdated(threadId: string): Promise<void> {
|
||||
const thread = await this.threadRepository.findOneBy({ id: threadId });
|
||||
if (!thread || thread.accessScope !== 'user' || !thread.ownerId) return;
|
||||
const data = { projectId: thread.projectId, agentId: thread.agentId, threadId };
|
||||
const userIds = await this.getRecipients(thread);
|
||||
if (userIds.length === 0) return;
|
||||
this.push.sendToUsers({ type: 'agentMessageQueueUpdated', data }, userIds);
|
||||
if (this.instanceSettings.isWorker || this.instanceSettings.isMultiMain) {
|
||||
await this.publisher.publishCommand({
|
||||
command: 'relay-agent-message-queue-update',
|
||||
payload: { data, userIds },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@OnPubSubEvent('relay-agent-message-queue-update', { instanceType: 'main' })
|
||||
handleQueueRelay({ data, userIds }: PubSubCommandMap['relay-agent-message-queue-update']): void {
|
||||
this.push.sendToUsers({ type: 'agentMessageQueueUpdated', data }, userIds);
|
||||
}
|
||||
|
||||
private async getRecipients(thread: AgentExecutionThread): Promise<string[]> {
|
||||
const userIds = await this.recipients.getProjectReaders(thread.projectId);
|
||||
return userIds.filter((id) => threadBelongsTo(thread, thread.projectId, thread.agentId, id));
|
||||
|
|
|
|||
|
|
@ -142,6 +142,12 @@ export class AgentMessageQueueConsumer {
|
|||
},
|
||||
controller,
|
||||
);
|
||||
sender.send({
|
||||
type: 'execution-started',
|
||||
executionId: admission.executionId,
|
||||
sessionId: thread.id,
|
||||
message: item.payload.message,
|
||||
});
|
||||
await this.chatExecutionService.settle(
|
||||
admission.executionId,
|
||||
async () => await this.consumePreview(claim, signal, sender.send),
|
||||
|
|
@ -224,8 +230,6 @@ export class AgentMessageQueueConsumer {
|
|||
admittedExecution: admission,
|
||||
abortSignal: signal,
|
||||
errorMode: 'forward',
|
||||
onExecutionStarted: (executionId, sessionId) =>
|
||||
send({ type: 'execution-started', executionId, sessionId }),
|
||||
onChunk: (chunk) => emitChunkEvents(chunk, send),
|
||||
});
|
||||
if (result.status === 'completed')
|
||||
|
|
|
|||
|
|
@ -1,7 +1,14 @@
|
|||
import type { AgentChatQueueResponse } from '@n8n/api-types';
|
||||
import { TransactionRunner, type OperationContext } from '@n8n/db';
|
||||
import { Service } from '@n8n/di';
|
||||
import { OperationalError, UserError } from 'n8n-workflow';
|
||||
|
||||
import { ConflictError } from '@/errors/response-errors/conflict.error';
|
||||
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
||||
import { NotFoundError } from '@/errors/response-errors/not-found.error';
|
||||
|
||||
import { AgentChatAttachmentService } from './agent-chat-attachment.service';
|
||||
import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster';
|
||||
import { AgentExecutionService, type StartExecutionParams } from './agent-execution.service';
|
||||
import type { AgentExecutionThread } from './entities/agent-execution-thread.entity';
|
||||
import type { AgentMessageQueue } from './entities/agent-message-queue.entity';
|
||||
|
|
@ -12,7 +19,7 @@ import { AgentExecutionThreadRepository } from './repositories/agent-execution-t
|
|||
import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository';
|
||||
import { AgentRepository } from './repositories/agent.repository';
|
||||
import type { AgentExecutionAdmission, AgentQueuedMessage } from './types/agent-queued-message';
|
||||
import type { AgentSessionMode } from './utils/agent-thread-access';
|
||||
import { canContinueThreadInPreview, type AgentSessionMode } from './utils/agent-thread-access';
|
||||
|
||||
export interface ClaimedAgentMessage {
|
||||
item: AgentMessageQueue;
|
||||
|
|
@ -35,6 +42,8 @@ export class AgentMessageQueueService {
|
|||
private readonly executionService: AgentExecutionService,
|
||||
private readonly checkpointStorage: N8NCheckpointStorage,
|
||||
private readonly agentRepository: AgentRepository,
|
||||
private readonly attachments: AgentChatAttachmentService,
|
||||
private readonly updates: AgentExecutionUpdateBroadcaster,
|
||||
) {}
|
||||
|
||||
/** Save a pending message. It is durably accepted when the transaction commits. */
|
||||
|
|
@ -70,10 +79,105 @@ export class AgentMessageQueueService {
|
|||
onInserted?.(inserted.id);
|
||||
return inserted;
|
||||
});
|
||||
if (payload.kind === 'preview') this.updates.notifyQueueUpdated(item.threadId);
|
||||
this.onAvailable?.(item.threadId);
|
||||
return item;
|
||||
}
|
||||
|
||||
async listPending(input: {
|
||||
projectId: string;
|
||||
agentId: string;
|
||||
threadId: string;
|
||||
userId: string;
|
||||
}): Promise<AgentChatQueueResponse> {
|
||||
const thread = await this.threadRepository.findOneBy({ id: input.threadId });
|
||||
// A client-created Preview session can have no accepted messages yet.
|
||||
if (!thread) return { items: [] };
|
||||
await this.assertPreviewAccess(thread, input);
|
||||
const items = await this.repository.listPending(thread.id);
|
||||
return {
|
||||
items: items
|
||||
.filter((item) => item.payload.kind === 'preview')
|
||||
.map((item) => ({
|
||||
id: item.id,
|
||||
message: item.payload.message,
|
||||
attachments: item.payload.attachments,
|
||||
createdAt: item.createdAt.toISOString(),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
async removePending(input: {
|
||||
projectId: string;
|
||||
agentId: string;
|
||||
threadId: string;
|
||||
userId: string;
|
||||
queueId: string;
|
||||
}): Promise<void> {
|
||||
const item = await this.txRunner.run({}, async (ctx) => {
|
||||
const thread = await this.threadRepository.lockById(input.threadId, ctx);
|
||||
if (!thread) throw new NotFoundError('Session not found');
|
||||
await this.assertPreviewAccess(thread, input, ctx);
|
||||
const item = await this.repository.findItem(thread.id, input.queueId, ctx);
|
||||
if (!item || item.payload.kind !== 'preview')
|
||||
throw new NotFoundError('Queued message not found');
|
||||
if (
|
||||
item.executionId !== null ||
|
||||
!(await this.repository.removePending(thread.id, item.id, ctx))
|
||||
) {
|
||||
throw new ConflictError('This message has already started');
|
||||
}
|
||||
return item;
|
||||
});
|
||||
this.updates.notifyQueueUpdated(item.threadId);
|
||||
await this.attachments.deleteByIds(item.payload.attachments?.map(({ id }) => id) ?? []);
|
||||
}
|
||||
|
||||
async updatePending(input: {
|
||||
projectId: string;
|
||||
agentId: string;
|
||||
threadId: string;
|
||||
userId: string;
|
||||
queueId: string;
|
||||
message: string;
|
||||
}): Promise<void> {
|
||||
await this.txRunner.run({}, async (ctx) => {
|
||||
const thread = await this.threadRepository.lockById(input.threadId, ctx);
|
||||
if (!thread) throw new NotFoundError('Session not found');
|
||||
await this.assertPreviewAccess(thread, input, ctx);
|
||||
const item = await this.repository.findItem(thread.id, input.queueId, ctx);
|
||||
if (!item || item.payload.kind !== 'preview')
|
||||
throw new NotFoundError('Queued message not found');
|
||||
if (item.executionId !== null) throw new ConflictError('This message has already started');
|
||||
const message = input.message.trim();
|
||||
if (!message && !item.payload.attachments?.length)
|
||||
throw new BadRequestError('A message or attachment is required');
|
||||
const updated = await this.repository.updatePendingPayload(
|
||||
thread.id,
|
||||
item.id,
|
||||
{ ...item.payload, message },
|
||||
ctx,
|
||||
);
|
||||
if (!updated) throw new ConflictError('This message has already started');
|
||||
});
|
||||
this.updates.notifyQueueUpdated(input.threadId);
|
||||
}
|
||||
|
||||
private async assertPreviewAccess(
|
||||
thread: AgentExecutionThread,
|
||||
input: { projectId: string; agentId: string; userId: string },
|
||||
ctx: OperationContext = {},
|
||||
): Promise<void> {
|
||||
const sources = await this.executionRepository.findFirstSourceByThreadIds([thread.id], ctx);
|
||||
if (
|
||||
thread.projectId !== input.projectId ||
|
||||
thread.agentId !== input.agentId ||
|
||||
!canContinueThreadInPreview(thread, input.userId, sources.get(thread.id))
|
||||
) {
|
||||
throw new NotFoundError('Session not found');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Give the oldest pending message exclusive use of the session for its execution.
|
||||
* Running work and valid suspended checkpoints block a claim.
|
||||
|
|
|
|||
|
|
@ -152,7 +152,10 @@ export class AgentExecutionRepository extends BaseRepository<AgentExecution> {
|
|||
*
|
||||
* Returns one row per thread from that thread's earliest matching run.
|
||||
*/
|
||||
async findFirstSourceByThreadIds(threadIds: string[]): Promise<Map<string, string>> {
|
||||
async findFirstSourceByThreadIds(
|
||||
threadIds: string[],
|
||||
ctx: OperationContext = {},
|
||||
): Promise<Map<string, string>> {
|
||||
if (threadIds.length === 0) return new Map();
|
||||
|
||||
// Correlated subquery: for each thread, pick the row with the smallest
|
||||
|
|
@ -160,7 +163,9 @@ export class AgentExecutionRepository extends BaseRepository<AgentExecution> {
|
|||
// so Postgres preserves their camelCase (it lowercases unquoted names),
|
||||
// and the table name is read from metadata so DB_TABLE_PREFIX is respected.
|
||||
const tableName = this.metadata.tablePath;
|
||||
const rows = await this.createQueryBuilder('e')
|
||||
const rows = await this.managerFor(ctx)
|
||||
.getRepository(AgentExecution)
|
||||
.createQueryBuilder('e')
|
||||
.select(['e."threadId" AS "threadId"', 'e."source" AS "source"'])
|
||||
.where('e."threadId" IN (:...threadIds)', { threadIds })
|
||||
.andWhere('e."source" IS NOT NULL')
|
||||
|
|
|
|||
|
|
@ -25,6 +25,37 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
|
|||
);
|
||||
}
|
||||
|
||||
async listPending(threadId: string) {
|
||||
return await this.find({ where: { threadId, executionId: IsNull() }, order: { id: 'ASC' } });
|
||||
}
|
||||
|
||||
async findItem(threadId: string, id: string, ctx: OperationContext) {
|
||||
return await this.managerFor(ctx).findOneBy(AgentMessageQueue, { threadId, id });
|
||||
}
|
||||
|
||||
async removePending(threadId: string, id: string, ctx: OperationContext) {
|
||||
const result = await this.managerFor(ctx).delete(AgentMessageQueue, {
|
||||
threadId,
|
||||
id,
|
||||
executionId: IsNull(),
|
||||
});
|
||||
return result.affected === 1;
|
||||
}
|
||||
|
||||
async updatePendingPayload(
|
||||
threadId: string,
|
||||
id: string,
|
||||
payload: AgentQueuedMessage,
|
||||
ctx: OperationContext,
|
||||
) {
|
||||
const result = await this.managerFor(ctx).update(
|
||||
AgentMessageQueue,
|
||||
{ threadId, id, executionId: IsNull() },
|
||||
{ payload },
|
||||
);
|
||||
return result.affected === 1;
|
||||
}
|
||||
|
||||
async findHead(threadId: string, ctx: OperationContext) {
|
||||
return await this.managerFor(ctx).findOne(AgentMessageQueue, {
|
||||
where: { threadId },
|
||||
|
|
|
|||
|
|
@ -47,6 +47,7 @@ export const IMMEDIATE_COMMANDS = new Set<PubSub.Command['command']>([
|
|||
'relay-agent-execution-update',
|
||||
'relay-agent-queued-chat',
|
||||
'cancel-agent-chat-execution',
|
||||
'relay-agent-message-queue-update',
|
||||
'relay-agent-background-tasks-update',
|
||||
'relay-agent-update',
|
||||
'resume-agent-workflow-tool',
|
||||
|
|
|
|||
|
|
@ -173,6 +173,7 @@ describe('Publisher', () => {
|
|||
'relay-agent-execution-update',
|
||||
'cancel-agent-chat-execution',
|
||||
'relay-agent-background-tasks-update',
|
||||
'relay-agent-message-queue-update',
|
||||
'relay-agent-update',
|
||||
'display-workflow-activation',
|
||||
'display-workflow-deactivation',
|
||||
|
|
|
|||
|
|
@ -147,6 +147,11 @@ export type PubSubCommandMap = {
|
|||
userId: string;
|
||||
};
|
||||
|
||||
'relay-agent-message-queue-update': {
|
||||
data: PushPayload<'agentMessageQueueUpdated'>;
|
||||
userIds: string[];
|
||||
};
|
||||
|
||||
'relay-agent-background-tasks-update': {
|
||||
data: PushPayload<'agentBackgroundTasksUpdated'>;
|
||||
userIds: string[];
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ export namespace PubSub {
|
|||
export type RelayAgentExecutionUpdate = ToCommand<'relay-agent-execution-update'>;
|
||||
export type RelayAgentQueuedChat = ToCommand<'relay-agent-queued-chat'>;
|
||||
export type CancelAgentChatExecution = ToCommand<'cancel-agent-chat-execution'>;
|
||||
export type RelayAgentMessageQueueUpdate = ToCommand<'relay-agent-message-queue-update'>;
|
||||
export type RelayAgentBackgroundJobsUpdate = ToCommand<'relay-agent-background-tasks-update'>;
|
||||
export type RelayAgentUpdate = ToCommand<'relay-agent-update'>;
|
||||
export type ResumeAgentWorkflowTool = ToCommand<'resume-agent-workflow-tool'>;
|
||||
|
|
@ -112,6 +113,7 @@ export namespace PubSub {
|
|||
| Commands.RelayAgentExecutionUpdate
|
||||
| Commands.RelayAgentQueuedChat
|
||||
| Commands.CancelAgentChatExecution
|
||||
| Commands.RelayAgentMessageQueueUpdate
|
||||
| Commands.RelayAgentBackgroundJobsUpdate
|
||||
| Commands.RelayAgentUpdate
|
||||
| Commands.ResumeAgentWorkflowTool
|
||||
|
|
|
|||
|
|
@ -178,6 +178,8 @@ describe('AgentExecutionRepository', () => {
|
|||
executionService,
|
||||
checkpointStorage,
|
||||
connection ? new AgentRepository(connection) : agentRepo,
|
||||
attachmentService,
|
||||
mock<AgentExecutionUpdateBroadcaster>(),
|
||||
);
|
||||
return {
|
||||
txRunner,
|
||||
|
|
@ -1058,6 +1060,157 @@ describe('AgentExecutionRepository', () => {
|
|||
await services.queue.settle(item.thread.id, item.admission.executionId);
|
||||
}
|
||||
|
||||
it('lists only pending Preview input and removes its attachments without affecting the active run', async () => {
|
||||
const services = recordingServices();
|
||||
const threadId = uuid();
|
||||
const target = { projectId, agentId, threadId, userId: owner.id };
|
||||
await services.queue.enqueue(input(threadId, 'active', 'new'));
|
||||
const active = await claim(services, threadId);
|
||||
const pendingInput = input(threadId, 'pending');
|
||||
pendingInput.payload.attachments = [
|
||||
{ id: 'pending-file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 },
|
||||
];
|
||||
const pending = await services.queue.enqueue(pendingInput);
|
||||
expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
|
||||
pending.id,
|
||||
]);
|
||||
await expect(
|
||||
services.queue.removePending({ ...target, queueId: active.item.id }),
|
||||
).rejects.toThrow('already started');
|
||||
await expect(
|
||||
services.queue.removePending({ ...target, userId: 'other-user', queueId: pending.id }),
|
||||
).rejects.toThrow('Session not found');
|
||||
await expect(
|
||||
services.queue.listPending({ ...target, agentId: 'other-agent' }),
|
||||
).rejects.toThrow('Session not found');
|
||||
expect(services.attachmentService.deleteByIds).not.toHaveBeenCalled();
|
||||
await services.queue.removePending({ ...target, queueId: pending.id });
|
||||
expect(await services.queue.listPending(target)).toEqual({ items: [] });
|
||||
expect(services.attachmentService.deleteByIds).toHaveBeenCalledWith(['pending-file']);
|
||||
expect((await repository.findOneByOrFail({ id: active.admission.executionId })).status).toBe(
|
||||
'running',
|
||||
);
|
||||
await finish(services, active);
|
||||
});
|
||||
|
||||
it('edits pending text without changing identity, attachments, or queue position', async () => {
|
||||
const services = recordingServices();
|
||||
const threadId = uuid();
|
||||
const data = input(threadId, 'original', 'new');
|
||||
data.payload.attachments = [
|
||||
{ id: 'file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 },
|
||||
];
|
||||
const first = await services.queue.enqueue(data);
|
||||
const second = await services.queue.enqueue(input(threadId, 'second'));
|
||||
const target = {
|
||||
projectId,
|
||||
agentId,
|
||||
threadId,
|
||||
userId: owner.id,
|
||||
queueId: first.id,
|
||||
message: 'updated',
|
||||
};
|
||||
await expect(
|
||||
services.queue.updatePending({ ...target, userId: 'other-user' }),
|
||||
).rejects.toThrow('Session not found');
|
||||
await expect(
|
||||
services.queue.updatePending({ ...target, agentId: 'other-agent' }),
|
||||
).rejects.toThrow('Session not found');
|
||||
await expect(
|
||||
services.queue.updatePending({ ...target, queueId: second.id, message: ' ' }),
|
||||
).rejects.toThrow('message or attachment');
|
||||
await services.queue.updatePending({ ...target, message: ' ' });
|
||||
expect((await services.queue.listPending(target)).items[0]).toMatchObject({
|
||||
message: '',
|
||||
attachments: data.payload.attachments,
|
||||
});
|
||||
await services.queue.updatePending(target);
|
||||
const stored = await services.queueRepository.findOneByOrFail({ id: first.id });
|
||||
expect(stored).toMatchObject({
|
||||
id: first.id,
|
||||
createdAt: first.createdAt,
|
||||
source: first.source,
|
||||
payload: { ...first.payload, message: 'updated' },
|
||||
});
|
||||
expect((await services.queue.listPending(target)).items.map(({ id }) => id)).toEqual([
|
||||
first.id,
|
||||
second.id,
|
||||
]);
|
||||
expect(services.attachmentService.deleteByIds).not.toHaveBeenCalled();
|
||||
const claimed = await claim(services, threadId);
|
||||
expect(claimed.item.payload.message).toBe('updated');
|
||||
await finish(services, claimed);
|
||||
await expect(services.queue.updatePending(target)).rejects.toThrow(
|
||||
'Queued message not found',
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
['claim', 'remove'],
|
||||
['remove', 'remove'],
|
||||
['claim', 'edit'],
|
||||
['edit', 'edit'],
|
||||
] as const)(
|
||||
'serializes concurrent connections when %s wins the claim-versus-%s race',
|
||||
async (winner, operation) => {
|
||||
const local = recordingServices();
|
||||
const remote = recordingServices(undefined, peer);
|
||||
const threadId = uuid();
|
||||
const item = await local.queue.enqueue(input(threadId, 'first', 'new'));
|
||||
const target = { projectId, agentId, threadId, userId: owner.id, queueId: item.id };
|
||||
const acquired = createDeferredPromise();
|
||||
const release = createDeferredPromise();
|
||||
const lock = local.threads.lockById.bind(local.threads);
|
||||
vi.spyOn(local.threads, 'lockById').mockImplementationOnce(async (...args) => {
|
||||
const thread = await lock(...args);
|
||||
acquired.resolve();
|
||||
await release.promise;
|
||||
return thread;
|
||||
});
|
||||
const observe = observePeerTransaction();
|
||||
const mutate = async (services: ReturnType<typeof recordingServices>) =>
|
||||
operation === 'edit'
|
||||
? await services.queue.updatePending({ ...target, message: 'edited' })
|
||||
: await services.queue.removePending(target);
|
||||
const first =
|
||||
winner === 'claim' ? local.queue.claimNext(threadId, async () => true) : mutate(local);
|
||||
await acquired.promise;
|
||||
const second =
|
||||
winner === 'claim' ? mutate(remote) : remote.queue.claimNext(threadId, async () => true);
|
||||
const results = Promise.allSettled([first, second]);
|
||||
try {
|
||||
await observe.started;
|
||||
} finally {
|
||||
release.resolve();
|
||||
observe.restore();
|
||||
}
|
||||
const settled = await results;
|
||||
expect(settled[0].status).toBe('fulfilled');
|
||||
if (winner === 'claim') {
|
||||
expect(settled[1]).toMatchObject({
|
||||
status: 'rejected',
|
||||
reason: { message: 'This message has already started' },
|
||||
});
|
||||
const claimed = await first;
|
||||
if (!claimed) throw new Error('Expected a claim');
|
||||
expect(await local.queueRepository.findOneByOrFail({ id: item.id })).toMatchObject({
|
||||
threadId,
|
||||
executionId: claimed.admission.executionId,
|
||||
});
|
||||
await finish(local, claimed);
|
||||
} else if (operation === 'edit') {
|
||||
const claimed = await second;
|
||||
if (!claimed) throw new Error('Expected a claim');
|
||||
expect(claimed.item.payload.message).toBe('edited');
|
||||
await finish(remote, claimed);
|
||||
} else {
|
||||
expect(settled[1]).toEqual({ status: 'fulfilled', value: null });
|
||||
expect(await repository.countBy({ threadId })).toBe(0);
|
||||
}
|
||||
expect(await local.queueRepository.countBy({ threadId })).toBe(0);
|
||||
},
|
||||
);
|
||||
|
||||
it('serializes acceptance and claims across connections while other sessions progress', async () => {
|
||||
const local = recordingServices();
|
||||
const remote = recordingServices(undefined, peer);
|
||||
|
|
|
|||
|
|
@ -7973,6 +7973,16 @@
|
|||
"agents.chat.loadHistory.error": "Could not load chat history",
|
||||
"agents.chat.clearHistory.error": "Could not clear chat history",
|
||||
"agents.chat.stop.error": "Couldn't stop the agent. Try again.",
|
||||
"agents.chat.queue.title": "{count} message up next | {count} messages up next",
|
||||
"agents.chat.queue.edit": "Edit queued message",
|
||||
"agents.chat.queue.save": "Save changes",
|
||||
"agents.chat.queue.cancelEdit": "Discard edits",
|
||||
"agents.chat.queue.editUnavailable": "This message has started or was removed. Your changes weren't applied.",
|
||||
"agents.chat.queue.editError": "Couldn't save your changes. Try again.",
|
||||
"agents.chat.queue.remove": "Remove queued message",
|
||||
"agents.chat.queue.loadError": "Couldn't load queued messages. Reopen the chat to try again.",
|
||||
"agents.chat.queue.removeError": "Couldn't remove the message. Check whether it has started and try again.",
|
||||
"agents.chat.queue.sendError": "Couldn't confirm the message was queued. Check the queue before sending again.",
|
||||
"agents.chat.backgroundTasks.resultsReceived": "Background task results received",
|
||||
"agents.chat.backgroundTasks.result": "{title} — {status}",
|
||||
"agents.chat.backgroundTasks.runningCount": "Running {count} background task | Running {count} background tasks",
|
||||
|
|
@ -8022,8 +8032,6 @@
|
|||
"agents.chat.approval.approved": "Approved",
|
||||
"agents.chat.approval.rejected": "Rejected",
|
||||
"agents.chat.approval.viewToolDetails": "View tool details",
|
||||
"agents.chat.approval.inputPlaceholder": "Approve or reject to continue",
|
||||
"agents.chat.waiting.inputPlaceholder": "Waiting on the card above to continue",
|
||||
"agents.chat.delegate.label": "Sub-agent · {name}",
|
||||
"agents.chat.delegate.labelFallback": "Sub-agent",
|
||||
"agents.chat.delegate.childSuspendUnsupported": "Sub-agent requested user input, which is not supported for delegated runs yet.",
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { computed, defineComponent, h, ref, nextTick } from 'vue';
|
|||
import { createMemoryHistory, createRouter } from 'vue-router';
|
||||
import { AGENT_SESSION_DETAIL_VIEW } from '../constants';
|
||||
import { APPROVAL_TOOL_NAME, N8N_CHAT_ACTION_TOOL_NAME, WAIT_TOOL_NAME } from '@n8n/api-types';
|
||||
import type { AgentBackgroundJobDto } from '@n8n/api-types';
|
||||
import type { AgentChatQueueItem, AgentBackgroundJobDto } from '@n8n/api-types';
|
||||
import type { ChatMessage } from '@/features/ai/shared/agentsChat/types';
|
||||
import AgentChatPanel from '../components/AgentChatPanel.vue';
|
||||
import AgentPreviewDock from '../components/AgentPreviewDock.vue';
|
||||
|
|
@ -23,6 +23,12 @@ const cancelAndSteerMock = vi.fn();
|
|||
const focusInputMock = vi.fn();
|
||||
const messagesMock = ref<ChatMessage[]>([]);
|
||||
const isStreamingMock = ref(false);
|
||||
const isSubmittingMock = ref(false);
|
||||
const isLoadingHistoryMock = ref(false);
|
||||
const trackSubmittedMessageMock = vi.fn();
|
||||
const queuedMessagesMock = ref<AgentChatQueueItem[]>([]);
|
||||
const removeQueuedMessageMock = vi.fn();
|
||||
const updateQueuedMessageMock = vi.fn();
|
||||
const isCancellingMock = ref(false);
|
||||
const backgroundJobsMock = ref<AgentBackgroundJobDto[]>([]);
|
||||
vi.mock('../composables/useAgentBackgroundJobs', () => ({
|
||||
|
|
@ -45,6 +51,7 @@ vi.mock('@n8n/i18n', () => {
|
|||
) => {
|
||||
const translations: Record<string, string> = {
|
||||
'agents.chat.input.placeholder.withAgent': `Message ${options?.interpolate?.agentName}…`,
|
||||
'agents.chat.queue.title': `${options?.interpolate?.count} ${options?.adjustToNumber === 1 ? 'message' : 'messages'} up next`,
|
||||
'agents.chat.misconfigured.issuesPrefix': 'Check:',
|
||||
'agents.chat.misconfigured.missing.tools': 'Tool configuration',
|
||||
'agents.chat.misconfigured.missing.mcpServers': 'MCP server',
|
||||
|
|
@ -81,7 +88,8 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({
|
|||
N8nLink: (await importOriginal<typeof import('@n8n/design-system')>()).N8nLink,
|
||||
useDropdownSearch: (await importOriginal<typeof import('@n8n/design-system')>())
|
||||
.useDropdownSearch,
|
||||
N8nButton: { template: '<button><slot /></button>' },
|
||||
N8nInput: (await importOriginal<typeof import('@n8n/design-system')>()).N8nInput,
|
||||
N8nButton: { template: '<button><slot name="icon" /><slot /></button>' },
|
||||
N8nCallout: { template: '<div><slot /><slot name="trailingContent" /></div>' },
|
||||
N8nDropdownMenu: { template: '<div><slot name="trigger" /></div>' },
|
||||
N8nHeading: { template: '<div><slot /></div>' },
|
||||
|
|
@ -91,12 +99,6 @@ vi.mock('@n8n/design-system', async (importOriginal) => ({
|
|||
template: '<button v-bind="$attrs" @click="$emit(\'click\')" />',
|
||||
},
|
||||
N8nText: { template: '<span><slot /></span>' },
|
||||
N8nSendStopButton: {
|
||||
name: 'N8nSendStopButton',
|
||||
props: ['streaming', 'stopButtonTestId'],
|
||||
emits: ['stop'],
|
||||
template: '<button :data-test-id="stopButtonTestId" @click="$emit(\'stop\')" />',
|
||||
},
|
||||
N8nTooltip: { template: '<div><slot /></div>' },
|
||||
TOOLTIP_DELAY_MS: 500,
|
||||
}));
|
||||
|
|
@ -149,7 +151,15 @@ vi.mock('@/features/ai/shared/components/ChatInputBase.vue', async () => {
|
|||
name: 'ChatInputBase',
|
||||
template:
|
||||
'<form data-testid="chat-input-stub" @submit.prevent="$emit(\'submit\')"><slot name="header" /><textarea ref="input" /><slot name="footer-start" /></form>',
|
||||
props: ['modelValue', 'placeholder', 'isStreaming', 'canSubmit', 'disabled', 'maxLength'],
|
||||
props: [
|
||||
'modelValue',
|
||||
'placeholder',
|
||||
'isStreaming',
|
||||
'showStopButton',
|
||||
'canSubmit',
|
||||
'disabled',
|
||||
'maxLength',
|
||||
],
|
||||
emits: ['submit', 'stop', 'update:modelValue', 'files-selected'],
|
||||
setup(_, { expose }) {
|
||||
const input = ref<HTMLTextAreaElement>();
|
||||
|
|
@ -184,6 +194,12 @@ vi.mock('../composables/useAgentChatStream', () => ({
|
|||
return {
|
||||
messages: messagesMock,
|
||||
isStreaming: isStreamingMock,
|
||||
isSubmitting: isSubmittingMock,
|
||||
isLoadingHistory: isLoadingHistoryMock,
|
||||
queuedMessages: queuedMessagesMock,
|
||||
removingQueueIds: ref(new Set()),
|
||||
removeQueuedMessage: removeQueuedMessageMock,
|
||||
updateQueuedMessage: updateQueuedMessageMock,
|
||||
isCancelling: isCancellingMock,
|
||||
messagingState: computed(() => (isStreamingMock.value ? 'receiving' : 'idle')),
|
||||
fatalError: fatalErrorMock,
|
||||
|
|
@ -200,7 +216,7 @@ vi.mock('../composables/useAgentChatStream', () => ({
|
|||
}));
|
||||
|
||||
vi.mock('../composables/useAgentTelemetry', () => ({
|
||||
useAgentTelemetry: () => ({ trackSubmittedMessage: vi.fn() }),
|
||||
useAgentTelemetry: () => ({ trackSubmittedMessage: trackSubmittedMessageMock }),
|
||||
}));
|
||||
|
||||
vi.mock('../composables/agentTelemetry.utils', () => ({
|
||||
|
|
@ -219,8 +235,12 @@ vi.mock('../composables/agentTelemetry.utils', () => ({
|
|||
describe('AgentChatPanel', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
updateQueuedMessageMock.mockReset().mockResolvedValue('updated');
|
||||
messagesMock.value = [];
|
||||
isStreamingMock.value = false;
|
||||
isSubmittingMock.value = false;
|
||||
isLoadingHistoryMock.value = false;
|
||||
queuedMessagesMock.value = [];
|
||||
isCancellingMock.value = false;
|
||||
backgroundJobsMock.value = [];
|
||||
fatalErrorMock.value = null;
|
||||
|
|
@ -262,6 +282,156 @@ describe('AgentChatPanel', () => {
|
|||
});
|
||||
}
|
||||
|
||||
it('keeps two pending messages in the composer below background tasks and removes them without adding conversation bubbles', async () => {
|
||||
queuedMessagesMock.value = [
|
||||
{ id: '1', message: 'Next message', createdAt: new Date().toISOString() },
|
||||
{
|
||||
id: '2',
|
||||
message: '',
|
||||
createdAt: new Date().toISOString(),
|
||||
attachments: [
|
||||
{ id: 'file-1', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 },
|
||||
],
|
||||
},
|
||||
];
|
||||
backgroundJobsMock.value = [
|
||||
{
|
||||
id: 'job-1',
|
||||
kind: 'subagent',
|
||||
title: 'Research',
|
||||
status: 'running',
|
||||
startedAt: new Date().toISOString(),
|
||||
},
|
||||
];
|
||||
isStreamingMock.value = true;
|
||||
const wrapper = mountPanel({ backgroundJobsActive: true });
|
||||
const composer = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
expect(composer.find('[data-testid="agent-message-queue"]').exists()).toBe(true);
|
||||
expect(composer.find('[data-testid="agent-background-jobs"]').exists()).toBe(false);
|
||||
expect(wrapper.find('[data-testid="agent-message-queue"] [aria-expanded]').exists()).toBe(
|
||||
false,
|
||||
);
|
||||
expect(
|
||||
wrapper.findAll('[data-testid="agent-queued-message"]').map((row) => row.text()),
|
||||
).toEqual(['Next message', 'notes.txt']);
|
||||
expect(wrapper.html().indexOf('agent-background-jobs')).toBeLessThan(
|
||||
wrapper.html().indexOf('agent-message-queue'),
|
||||
);
|
||||
expect(messagesMock.value).toEqual([]);
|
||||
const removeButtons = wrapper.findAll('[aria-label="agents.chat.queue.remove"]');
|
||||
for (const button of removeButtons) {
|
||||
expect(button.findComponent({ name: 'N8nIcon' }).props('icon')).toBe('trash-2');
|
||||
}
|
||||
await removeButtons[1].trigger('click');
|
||||
expect(removeQueuedMessageMock).toHaveBeenCalledWith('2');
|
||||
queuedMessagesMock.value = [];
|
||||
await nextTick();
|
||||
expect(wrapper.find('[data-testid="agent-message-queue"]').exists()).toBe(false);
|
||||
expect(wrapper.find('[data-testid="agent-background-jobs"]').exists()).toBe(true);
|
||||
wrapper.unmount();
|
||||
});
|
||||
|
||||
it('collapses only the third and later pending messages as the queue grows and shrinks', async () => {
|
||||
const items = [1, 2, 3, 4].map((id) => ({
|
||||
id: String(id),
|
||||
message: `Message ${id}`,
|
||||
createdAt: '2026-09-24T12:00:00.000Z',
|
||||
}));
|
||||
queuedMessagesMock.value = items.slice(0, 2);
|
||||
const wrapper = mountPanel();
|
||||
const queue = wrapper.get('[data-testid="agent-message-queue"]');
|
||||
expect(queue.find('[aria-expanded]').exists()).toBe(false);
|
||||
|
||||
queuedMessagesMock.value = items.slice(0, 3);
|
||||
await nextTick();
|
||||
const toggle = queue.get('button[aria-expanded]');
|
||||
expect(toggle.attributes('aria-expanded')).toBe('false');
|
||||
expect(toggle.text()).toBe('1 message up next');
|
||||
expect(queue.findAll('li').map((row) => row.text())).toEqual(['Message 1', 'Message 2']);
|
||||
|
||||
queuedMessagesMock.value = items;
|
||||
await nextTick();
|
||||
expect(toggle.text()).toBe('2 messages up next');
|
||||
expect(queue.findAll('li').map((row) => row.text())).toEqual(['Message 1', 'Message 2']);
|
||||
await toggle.trigger('click');
|
||||
expect(queue.findAll('li').map((row) => row.text())).toEqual([
|
||||
'Message 1',
|
||||
'Message 2',
|
||||
'Message 3',
|
||||
'Message 4',
|
||||
]);
|
||||
await toggle.trigger('click');
|
||||
expect(queue.findAll('li').map((row) => row.text())).toEqual(['Message 1', 'Message 2']);
|
||||
|
||||
queuedMessagesMock.value = items.slice(1, 3);
|
||||
await nextTick();
|
||||
expect(queue.find('[aria-expanded]').exists()).toBe(false);
|
||||
expect(queue.findAll('li').map((row) => row.text())).toEqual(['Message 2', 'Message 3']);
|
||||
wrapper.unmount();
|
||||
});
|
||||
|
||||
it('edits queued text inline while preserving attachments and the composer draft', async () => {
|
||||
queuedMessagesMock.value = [
|
||||
{
|
||||
id: '1',
|
||||
message: 'original',
|
||||
createdAt: new Date().toISOString(),
|
||||
attachments: [{ id: 'file', fileName: 'notes.txt', mimeType: 'text/plain', sizeBytes: 5 }],
|
||||
},
|
||||
];
|
||||
const wrapper = mountPanel();
|
||||
const composer = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
composer.vm.$emit('update:modelValue', 'next draft');
|
||||
await wrapper.get('[aria-label="agents.chat.queue.edit"]').trigger('click');
|
||||
let editor = wrapper.get('textarea[aria-label="agents.chat.queue.edit"]');
|
||||
await editor.setValue('changed');
|
||||
await editor.trigger('keydown', { key: 'Escape' });
|
||||
expect(wrapper.find('textarea[aria-label="agents.chat.queue.edit"]').exists()).toBe(false);
|
||||
expect(updateQueuedMessageMock).not.toHaveBeenCalled();
|
||||
await wrapper.get('[aria-label="agents.chat.queue.edit"]').trigger('click');
|
||||
editor = wrapper.get('textarea[aria-label="agents.chat.queue.edit"]');
|
||||
await editor.setValue('updated text');
|
||||
await editor.trigger('keydown', { key: 'Enter', shiftKey: true });
|
||||
expect(updateQueuedMessageMock).not.toHaveBeenCalled();
|
||||
await editor.trigger('keydown', { key: 'Enter', isComposing: true });
|
||||
expect(updateQueuedMessageMock).not.toHaveBeenCalled();
|
||||
await editor.trigger('keydown', { key: 'Enter' });
|
||||
await flushPromises();
|
||||
expect(updateQueuedMessageMock).toHaveBeenCalledExactlyOnceWith('1', 'updated text');
|
||||
expect(composer.props('modelValue')).toBe('next draft');
|
||||
expect(wrapper.get('[data-testid="agent-queued-message"]').text()).toContain('notes.txt');
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
wrapper.unmount();
|
||||
});
|
||||
|
||||
it.each(['notification', 'conflict'])(
|
||||
'preserves edits when a message starts before saving (%s)',
|
||||
async (source) => {
|
||||
queuedMessagesMock.value = [
|
||||
{ id: '1', message: 'original', createdAt: new Date().toISOString() },
|
||||
];
|
||||
const wrapper = mountPanel();
|
||||
await wrapper.get('[aria-label="agents.chat.queue.edit"]').trigger('click');
|
||||
const editor = wrapper.get('textarea[aria-label="agents.chat.queue.edit"]');
|
||||
await editor.setValue('unsaved text');
|
||||
if (source === 'notification') queuedMessagesMock.value = [];
|
||||
else {
|
||||
updateQueuedMessageMock.mockResolvedValueOnce('unavailable');
|
||||
await editor.trigger('keydown', { key: 'Enter' });
|
||||
}
|
||||
await flushPromises();
|
||||
expect(editor.element).toHaveProperty('value', 'unsaved text');
|
||||
expect(editor.attributes('readonly')).toBeDefined();
|
||||
expect(
|
||||
wrapper.get('[aria-label="agents.chat.queue.save"]').attributes('disabled'),
|
||||
).toBeDefined();
|
||||
expect(wrapper.text()).toContain('agents.chat.queue.editUnavailable');
|
||||
await wrapper.get('[aria-label="agents.chat.queue.cancelEdit"]').trigger('click');
|
||||
expect(wrapper.find('textarea[aria-label="agents.chat.queue.edit"]').exists()).toBe(false);
|
||||
wrapper.unmount();
|
||||
},
|
||||
);
|
||||
|
||||
describe('background task panel', () => {
|
||||
afterEach(() => vi.useRealTimers());
|
||||
const job: AgentBackgroundJobDto = {
|
||||
|
|
@ -380,7 +550,7 @@ describe('AgentChatPanel', () => {
|
|||
expect(wrapper.find('[data-testid="agent-background-jobs"]').exists()).toBe(false);
|
||||
backgroundJobsMock.value = [job, { ...job, id: 'job-2', title: 'Check tickets' }];
|
||||
await flushPromises();
|
||||
const panel = wrapper.get('[data-testid="chat-input"] [data-testid="agent-background-jobs"]');
|
||||
const panel = wrapper.get('[data-testid="agent-background-jobs"]');
|
||||
const trigger = panel.get('button');
|
||||
expect(trigger.text()).toContain('Running 2 background tasks');
|
||||
expect(trigger.attributes('aria-expanded')).toBe('false');
|
||||
|
|
@ -637,8 +807,7 @@ describe('AgentChatPanel', () => {
|
|||
|
||||
/**
|
||||
* A non-approval interactive card (`chat_action`) — these put the chat
|
||||
* input into cancel-and-steer mode rather than blocking it outright,
|
||||
* unlike an open approval card.
|
||||
* input into cancel-and-steer mode. Approval cards keep ordinary input queued.
|
||||
*/
|
||||
function openInteractiveMessage(): ChatMessage {
|
||||
return {
|
||||
|
|
@ -690,26 +859,62 @@ describe('AgentChatPanel', () => {
|
|||
expect(events).toEqual(['beforeSend', 'sendMessage']);
|
||||
});
|
||||
|
||||
it('queues an outside message until the current stream finishes', async () => {
|
||||
it('retains an outside prompt during initial loading and consumes it only after acceptance', async () => {
|
||||
isLoadingHistoryMock.value = true;
|
||||
isStreamingMock.value = true;
|
||||
const response = Promise.withResolvers<'sent'>();
|
||||
sendMessageMock.mockReturnValueOnce(response.promise);
|
||||
const wrapper = mountPanel();
|
||||
const input = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
expect(input.props('showStopButton')).toBe(false);
|
||||
(
|
||||
wrapper.vm as unknown as { sendMessageFromOutside: (message: string) => void }
|
||||
).sendMessageFromOutside('Test this task');
|
||||
await flushPromises();
|
||||
expect(input.props('canSubmit')).toBe(false);
|
||||
expect(input.props('showStopButton')).toBe(false);
|
||||
expect(input.props('modelValue')).toBe('Test this task');
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(trackSubmittedMessageMock).not.toHaveBeenCalled();
|
||||
expect(wrapper.emitted('initial-consumed')).toBeUndefined();
|
||||
|
||||
isLoadingHistoryMock.value = false;
|
||||
await flushPromises();
|
||||
expect(input.props('showStopButton')).toBe(false);
|
||||
expect(sendMessageMock).toHaveBeenCalledExactlyOnceWith(
|
||||
'Test this task',
|
||||
undefined,
|
||||
expect.any(Function),
|
||||
);
|
||||
expect(wrapper.emitted('initial-consumed')).toBeUndefined();
|
||||
expect(trackSubmittedMessageMock).not.toHaveBeenCalled();
|
||||
sendMessageMock.mock.lastCall?.[2]?.();
|
||||
response.resolve('sent');
|
||||
await flushPromises();
|
||||
expect(input.props('modelValue')).toBe('');
|
||||
expect(input.props('showStopButton')).toBe(true);
|
||||
expect(wrapper.emitted('initial-consumed')).toEqual([[]]);
|
||||
expect(trackSubmittedMessageMock).toHaveBeenCalledOnce();
|
||||
wrapper.unmount();
|
||||
});
|
||||
|
||||
it('submits an outside message while the current stream runs', async () => {
|
||||
const response = Promise.withResolvers<'sent'>();
|
||||
sendMessageMock.mockReturnValueOnce(response.promise);
|
||||
isStreamingMock.value = true;
|
||||
const wrapper = mountPanel();
|
||||
const input = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
expect(input.props('showStopButton')).toBe(true);
|
||||
|
||||
(
|
||||
wrapper.vm as unknown as { sendMessageFromOutside: (message: string) => void }
|
||||
).sendMessageFromOutside('Test these instructions');
|
||||
await flushPromises();
|
||||
|
||||
expect(wrapper.findComponent({ name: 'ChatInputBase' }).props('modelValue')).toBe(
|
||||
'Test these instructions',
|
||||
);
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(input.props('modelValue')).toBe('Test these instructions');
|
||||
expect(input.props('showStopButton')).toBe(false);
|
||||
expect(wrapper.emitted('initial-consumed')).toBeUndefined();
|
||||
|
||||
isStreamingMock.value = false;
|
||||
await flushPromises();
|
||||
|
||||
expect(sendMessageMock).toHaveBeenCalledExactlyOnceWith(
|
||||
'Test these instructions',
|
||||
undefined,
|
||||
|
|
@ -717,6 +922,7 @@ describe('AgentChatPanel', () => {
|
|||
);
|
||||
sendMessageMock.mock.lastCall?.[2]?.();
|
||||
await nextTick();
|
||||
expect(input.props('showStopButton')).toBe(true);
|
||||
expect(wrapper.emitted('initial-consumed')).toEqual([[]]);
|
||||
wrapper.unmount();
|
||||
response.resolve('sent');
|
||||
|
|
@ -796,7 +1002,7 @@ describe('AgentChatPanel', () => {
|
|||
it.each([
|
||||
{ state: 'streaming', busy: isStreamingMock },
|
||||
{ state: 'cancellation', busy: isCancellingMock },
|
||||
])('keeps the draft when $state starts during telemetry preparation', async ({ busy }) => {
|
||||
])('submits when $state starts during telemetry preparation', async ({ busy }) => {
|
||||
const fingerprint = Promise.withResolvers<AgentConfigFingerprint>();
|
||||
vi.mocked(buildAgentConfigFingerprint).mockReturnValueOnce(fingerprint.promise);
|
||||
const wrapper = mountPanel();
|
||||
|
|
@ -821,15 +1027,6 @@ describe('AgentChatPanel', () => {
|
|||
config_version: 'test-version',
|
||||
});
|
||||
await flushPromises();
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(chatInput.props('modelValue')).toBe(draft);
|
||||
expect(chatInput.props('canSubmit')).toBe(false);
|
||||
|
||||
busy.value = false;
|
||||
await flushPromises();
|
||||
expect(chatInput.props('canSubmit')).toBe(true);
|
||||
chatInput.vm.$emit('submit');
|
||||
await flushPromises();
|
||||
expect(sendMessageMock).toHaveBeenCalledExactlyOnceWith(
|
||||
draft.trim(),
|
||||
[file],
|
||||
|
|
@ -841,7 +1038,7 @@ describe('AgentChatPanel', () => {
|
|||
wrapper.unmount();
|
||||
});
|
||||
|
||||
it('keeps the draft while suspended-run cancellation is pending', async () => {
|
||||
it('submits while suspended-run cancellation is pending', async () => {
|
||||
isCancellingMock.value = true;
|
||||
const wrapper = mountPanel();
|
||||
|
||||
|
|
@ -853,7 +1050,11 @@ describe('AgentChatPanel', () => {
|
|||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
expect(chatInput.props('modelValue')).toBe('keep this draft');
|
||||
expect(chatInput.props('disabled')).toBe(false);
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(sendMessageMock).toHaveBeenCalledWith(
|
||||
'keep this draft',
|
||||
undefined,
|
||||
expect.any(Function),
|
||||
);
|
||||
wrapper.unmount();
|
||||
});
|
||||
|
||||
|
|
@ -862,7 +1063,7 @@ describe('AgentChatPanel', () => {
|
|||
async (outcome) => {
|
||||
const response = Promise.withResolvers<'sent' | 'busy'>();
|
||||
sendMessageMock.mockImplementationOnce(() => {
|
||||
isStreamingMock.value = true;
|
||||
isSubmittingMock.value = true;
|
||||
return response.promise;
|
||||
});
|
||||
const wrapper = mountPanel();
|
||||
|
|
@ -881,16 +1082,12 @@ describe('AgentChatPanel', () => {
|
|||
{ id: 'snapshot', role: 'assistant', content: 'progress', status: 'streaming' },
|
||||
];
|
||||
if (outcome === 'sent') sendMessageMock.mock.lastCall?.[2]?.();
|
||||
isSubmittingMock.value = false;
|
||||
response.resolve(outcome);
|
||||
await flushPromises();
|
||||
expect(input.props('modelValue')).toBe('edited draft');
|
||||
input.vm.$emit('submit');
|
||||
await flushPromises();
|
||||
expect(sendMessageMock).toHaveBeenCalledOnce();
|
||||
isStreamingMock.value = false;
|
||||
await flushPromises();
|
||||
input.vm.$emit('submit');
|
||||
await flushPromises();
|
||||
expect(sendMessageMock).toHaveBeenLastCalledWith(
|
||||
'edited draft',
|
||||
outcome === 'busy' ? [file, nextFile] : [nextFile],
|
||||
|
|
@ -936,7 +1133,7 @@ describe('AgentChatPanel', () => {
|
|||
expect(wrapper.findComponent({ name: 'ChatInputBase' }).props('modelValue')).toBe(
|
||||
'keep this direction',
|
||||
);
|
||||
expect(wrapper.emitted('initial-consumed')).toEqual([[]]);
|
||||
expect(wrapper.emitted('initial-consumed')).toBeUndefined();
|
||||
isStreamingMock.value = true;
|
||||
await nextTick();
|
||||
isStreamingMock.value = false;
|
||||
|
|
@ -977,16 +1174,23 @@ describe('AgentChatPanel', () => {
|
|||
expect(chatInput.props('disabled')).toBe(false);
|
||||
});
|
||||
|
||||
it('shows send and stop controls while an interactive question is unresolved', async () => {
|
||||
it('switches between Stop and Send as a question reply changes', async () => {
|
||||
messagesMock.value = [openInteractiveMessage()];
|
||||
|
||||
const wrapper = mountPanel();
|
||||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
|
||||
expect(chatInput.props('isStreaming')).toBe(false);
|
||||
const stopButton = wrapper.find('[data-test-id="agent-chat-suspended-stop-button"]');
|
||||
expect(stopButton.exists()).toBe(true);
|
||||
await stopButton.trigger('click');
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
chatInput.vm.$emit('update:modelValue', 'My answer');
|
||||
await nextTick();
|
||||
expect(chatInput.props('showStopButton')).toBe(false);
|
||||
expect(chatInput.props('canSubmit')).toBe(true);
|
||||
chatInput.vm.$emit('update:modelValue', ' ');
|
||||
await nextTick();
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
expect(chatInput.props('canSubmit')).toBe(false);
|
||||
chatInput.vm.$emit('stop');
|
||||
await flushPromises();
|
||||
expect(stopGeneratingMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
|
@ -994,7 +1198,7 @@ describe('AgentChatPanel', () => {
|
|||
/**
|
||||
* A workflow tool parked on a Wait node. It renders a card, but nobody is
|
||||
* being asked anything — the workflow resumes it — so the input must be
|
||||
* blocked rather than routed into cancel-and-steer.
|
||||
* queued rather than routed into cancel-and-steer.
|
||||
*/
|
||||
function openWaitMessage(): ChatMessage {
|
||||
return {
|
||||
|
|
@ -1019,18 +1223,19 @@ describe('AgentChatPanel', () => {
|
|||
};
|
||||
}
|
||||
|
||||
it('blocks the chat input while a waiting card is open', () => {
|
||||
it('allows queue submissions while a waiting card is open', () => {
|
||||
messagesMock.value = [openWaitMessage()];
|
||||
|
||||
const wrapper = mountPanel();
|
||||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
|
||||
expect(chatInput.props('disabled')).toBe(true);
|
||||
expect(chatInput.props('isStreaming')).toBe(true);
|
||||
expect(chatInput.props('placeholder')).toBe('agents.chat.waiting.inputPlaceholder');
|
||||
expect(chatInput.props('disabled')).toBe(false);
|
||||
expect(chatInput.props('isStreaming')).toBe(false);
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
expect(chatInput.props('placeholder')).toBe('Message Agent…');
|
||||
});
|
||||
|
||||
it('neither sends nor steers while a waiting card is open', async () => {
|
||||
it('queues ordinary messages without steering a waiting card', async () => {
|
||||
messagesMock.value = [openWaitMessage()];
|
||||
|
||||
const wrapper = mountPanel();
|
||||
|
|
@ -1040,7 +1245,7 @@ describe('AgentChatPanel', () => {
|
|||
).sendMessageFromOutside('any news?');
|
||||
await flushPromises();
|
||||
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(sendMessageMock).toHaveBeenCalledWith('any news?', undefined, expect.any(Function));
|
||||
expect(cancelAndSteerMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
|
@ -1143,7 +1348,7 @@ describe('AgentChatPanel', () => {
|
|||
|
||||
// The run is parked and would restart from a context with the pending tool
|
||||
// call stripped out, so the model would call the same tool again.
|
||||
it('blocks the chat input for a suspension with no card', () => {
|
||||
it('allows queue submissions for a suspension with no card', () => {
|
||||
messagesMock.value = [
|
||||
{
|
||||
id: 'assistant-1',
|
||||
|
|
@ -1158,10 +1363,10 @@ describe('AgentChatPanel', () => {
|
|||
const wrapper = mountPanel();
|
||||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
|
||||
expect(chatInput.props('disabled')).toBe(true);
|
||||
expect(chatInput.props('disabled')).toBe(false);
|
||||
});
|
||||
|
||||
it('keeps the stop control available for a non-card suspension', () => {
|
||||
it('switches from Stop to Send for an attachment-only draft during a non-card suspension', async () => {
|
||||
messagesMock.value = [
|
||||
{
|
||||
id: 'assistant-1',
|
||||
|
|
@ -1181,7 +1386,19 @@ describe('AgentChatPanel', () => {
|
|||
const wrapper = mountPanel();
|
||||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
|
||||
expect(chatInput.props('isStreaming')).toBe(true);
|
||||
expect(chatInput.props('isStreaming')).toBe(false);
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
const file = new File(['notes'], 'notes.txt', { type: 'text/plain' });
|
||||
chatInput.vm.$emit('files-selected', [file]);
|
||||
await nextTick();
|
||||
expect(chatInput.props('showStopButton')).toBe(false);
|
||||
expect(chatInput.props('canSubmit')).toBe(true);
|
||||
chatInput.vm.$emit('submit');
|
||||
await flushPromises();
|
||||
expect(sendMessageMock).toHaveBeenCalledWith('', [file], expect.any(Function));
|
||||
sendMessageMock.mock.lastCall?.[2]?.();
|
||||
await nextTick();
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
});
|
||||
|
||||
it('shows stop while tool calls are in-flight even when the stream ended (desync)', () => {
|
||||
|
|
@ -1206,7 +1423,8 @@ describe('AgentChatPanel', () => {
|
|||
const wrapper = mountPanel();
|
||||
const chatInput = wrapper.findComponent({ name: 'ChatInputBase' });
|
||||
|
||||
expect(chatInput.props('isStreaming')).toBe(true);
|
||||
expect(chatInput.props('isStreaming')).toBe(false);
|
||||
expect(chatInput.props('showStopButton')).toBe(true);
|
||||
});
|
||||
|
||||
it('does not apply a build-specific character limit', () => {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,9 @@ import { getFullApiResponse, makeRestApiRequest } from '@n8n/rest-api-client';
|
|||
import {
|
||||
cancelAgentChatExecution,
|
||||
getAgentBackgroundJobs,
|
||||
getAgentChatQueue,
|
||||
removeAgentQueuedMessage,
|
||||
updateAgentQueuedMessage,
|
||||
getChatMessages,
|
||||
listAgents,
|
||||
listAgentsPage,
|
||||
|
|
@ -92,6 +95,20 @@ describe('useAgentApi', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('encodes the queue route identifiers for listing, editing, and removal', async () => {
|
||||
const args = [restApiContext, 'project/1', 'agent/1', 'agent:chat#1'] as const;
|
||||
await getAgentChatQueue(...args);
|
||||
await updateAgentQueuedMessage(...args, 'queue/1', { message: 'edited' });
|
||||
await removeAgentQueuedMessage(...args, 'queue/1');
|
||||
|
||||
const path = '/projects/project%2F1/agents/v2/agent%2F1/chat/agent%3Achat%231/queue';
|
||||
expect(vi.mocked(makeRestApiRequest).mock.calls).toEqual([
|
||||
[restApiContext, 'GET', path],
|
||||
[restApiContext, 'PATCH', `${path}/queue%2F1`, { message: 'edited' }],
|
||||
[restApiContext, 'DELETE', `${path}/queue%2F1`],
|
||||
]);
|
||||
});
|
||||
|
||||
describe('getChatMessages', () => {
|
||||
it('percent-encodes the thread id so a rotated session id survives as a URL, not a fragment', async () => {
|
||||
vi.mocked(makeRestApiRequest).mockResolvedValueOnce({ messages: [] });
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -169,3 +169,23 @@ skill, What the skill does, and Allowed tools. Use `Save skill` to confirm.
|
|||
Add a section when an Agent-specific pattern applies to two or more Agent
|
||||
surfaces. Keep implementation details with the owning pattern. Do not duplicate
|
||||
global Design System guidance.
|
||||
|
||||
## Preview composer queue
|
||||
|
||||
Use one action on the right. Show Stop when a turn can be stopped and the
|
||||
composer has no text or attachments. Otherwise, show Send. Keep file and voice
|
||||
input available during a turn.
|
||||
Stack the background task card above the composer. Attach pending messages to
|
||||
the top of the composer. Use a subtle background and dividers between messages.
|
||||
Use muted gray for queue text and icons. Use the same gray for all queue icons.
|
||||
Give text more contrast than icons. Keep text contrast at least 4.5:1.
|
||||
Use `2xs` text and `large` icons. Keep the action targets at least 24 by 24 pixels.
|
||||
Keep the first two messages visible.
|
||||
Put the third and later messages in a collapsed activity group. Show the number
|
||||
of additional pending messages in its header. Keep messages in queue order when
|
||||
expanded.
|
||||
Keep pending messages out of the conversation until processing starts. Give each
|
||||
message a Remove action. Hide an empty queue section. Removal discards the
|
||||
message. It does not restore the composer draft.
|
||||
|
||||
Edit queued text in place. Use compact Save and Cancel icon actions. Enter saves, Shift+Enter adds a line, and Escape cancels. Keep attachments unchanged. Do not pause the queue during editing. If the message starts, disable Save and retain the draft until the user dismisses it.
|
||||
|
|
|
|||
|
|
@ -11,15 +11,17 @@ import {
|
|||
} from 'vue';
|
||||
import {
|
||||
N8nAiActivityStepGroup,
|
||||
N8nButton,
|
||||
N8nCallout,
|
||||
N8nIcon,
|
||||
N8nIconButton,
|
||||
N8nInput,
|
||||
N8nLink,
|
||||
N8nSendStopButton,
|
||||
N8nTooltip,
|
||||
} from '@n8n/design-system';
|
||||
import { useDocumentVisibility, useIntervalFn } from '@vueuse/core';
|
||||
import { createReusableTemplate, useDocumentVisibility, useIntervalFn } from '@vueuse/core';
|
||||
import { useI18n } from '@n8n/i18n';
|
||||
import {
|
||||
type AgentChatQueueItem,
|
||||
APPROVAL_TOOL_NAME,
|
||||
WAIT_TOOL_NAME,
|
||||
MAX_AGENT_CHAT_ATTACHMENT_SIZE_BYTES,
|
||||
|
|
@ -92,6 +94,12 @@ const toast = useToast();
|
|||
|
||||
const {
|
||||
messages,
|
||||
queuedMessages,
|
||||
removingQueueIds,
|
||||
removeQueuedMessage,
|
||||
updateQueuedMessage,
|
||||
isSubmitting,
|
||||
isLoadingHistory,
|
||||
isStreaming,
|
||||
refresh,
|
||||
isCancelling,
|
||||
|
|
@ -119,6 +127,62 @@ const {
|
|||
onSessionCreated: (sessionId) => emit('session-created', sessionId),
|
||||
});
|
||||
|
||||
const queueEdit = ref<{
|
||||
item: AgentChatQueueItem;
|
||||
text: string;
|
||||
unavailable: boolean;
|
||||
saving: boolean;
|
||||
}>();
|
||||
const queueRows = computed(() => {
|
||||
const edit = queueEdit.value;
|
||||
if (edit && !queuedMessages.value.some((item) => item.id === edit.item.id)) {
|
||||
return [...queuedMessages.value, edit.item];
|
||||
}
|
||||
return queuedMessages.value;
|
||||
});
|
||||
const [DefineQueueList, QueueList] = createReusableTemplate<{ items: AgentChatQueueItem[] }>({
|
||||
inheritAttrs: false,
|
||||
});
|
||||
const canSaveQueueEdit = computed(() => {
|
||||
const edit = queueEdit.value;
|
||||
return (
|
||||
edit &&
|
||||
!edit.saving &&
|
||||
!edit.unavailable &&
|
||||
(edit.text.trim().length > 0 || !!edit.item.attachments?.length)
|
||||
);
|
||||
});
|
||||
watch(queuedMessages, (items) => {
|
||||
if (queueEdit.value && !items.some((item) => item.id === queueEdit.value?.item.id)) {
|
||||
queueEdit.value.unavailable = true;
|
||||
}
|
||||
});
|
||||
function startQueueEdit(item: AgentChatQueueItem) {
|
||||
queueEdit.value = { item, text: item.message, unavailable: false, saving: false };
|
||||
}
|
||||
async function saveQueueEdit() {
|
||||
const edit = queueEdit.value;
|
||||
if (!edit || !canSaveQueueEdit.value) return;
|
||||
edit.saving = true;
|
||||
const result = await updateQueuedMessage(edit.item.id, edit.text);
|
||||
if (queueEdit.value !== edit) return;
|
||||
edit.saving = false;
|
||||
if (result === 'updated') queueEdit.value = undefined;
|
||||
else if (result === 'unavailable') edit.unavailable = true;
|
||||
}
|
||||
function onQueueEditKeydown(event: KeyboardEvent) {
|
||||
if (event.isComposing) return;
|
||||
if (event.key === 'Escape') {
|
||||
event.preventDefault();
|
||||
event.stopPropagation();
|
||||
if (!queueEdit.value?.saving) queueEdit.value = undefined;
|
||||
} else if (event.key === 'Enter' && !event.shiftKey) {
|
||||
event.preventDefault();
|
||||
event.stopPropagation();
|
||||
void saveQueueEdit();
|
||||
}
|
||||
}
|
||||
|
||||
const { jobs: backgroundJobs } = useAgentBackgroundJobs({
|
||||
projectId: () => props.projectId,
|
||||
agentId: () => props.agentId,
|
||||
|
|
@ -330,6 +394,9 @@ const inputText = computed<string>({
|
|||
}
|
||||
},
|
||||
});
|
||||
const hasDraft = computed(
|
||||
() => inputText.value.trim().length > 0 || attachedFiles.value.length > 0,
|
||||
);
|
||||
const isPreparingToSend = ref(false);
|
||||
let disposed = false;
|
||||
let queuedExternalMessage: string | undefined;
|
||||
|
|
@ -389,24 +456,8 @@ const hasOpenSuspension = computed(
|
|||
(toolCall) => toolCall.state === TOOL_CALL_STATE.SUSPENDED && toolCall.runId,
|
||||
) ?? false,
|
||||
);
|
||||
/**
|
||||
* A parked run owns the conversation: sending now would start a second run
|
||||
* whose context has the pending tool call stripped out, so the model would
|
||||
* re-invoke the same tool. Only an open question is exempt — answering or
|
||||
* steering it resumes the same run. Stop stays available either way.
|
||||
*/
|
||||
const inputBlockedBySuspension = computed(
|
||||
() =>
|
||||
hasOpenApproval.value ||
|
||||
hasOpenWaitCard.value ||
|
||||
(hasOpenSuspension.value && !hasOpenInteractiveQuestion.value),
|
||||
);
|
||||
const isSubmissionBlocked = computed(
|
||||
() =>
|
||||
isStreaming.value ||
|
||||
isCancelling.value ||
|
||||
isPreparingToSend.value ||
|
||||
inputBlockedBySuspension.value,
|
||||
() => isPreparingToSend.value || isSubmitting.value || isLoadingHistory.value,
|
||||
);
|
||||
// Tools still pending/running after the stream ended (desync): the backend
|
||||
// finished but their terminal events never arrived. Surfacing Stop here lets
|
||||
|
|
@ -419,24 +470,18 @@ const hasInFlightToolCalls = computed(() =>
|
|||
),
|
||||
),
|
||||
);
|
||||
const showSuspensionStopAlongsideSend = computed(
|
||||
() => hasOpenInteractiveQuestion.value && !isStreaming.value && !isCancelling.value,
|
||||
);
|
||||
const showStopAsPrimaryAction = computed(
|
||||
const showStop = computed(
|
||||
() =>
|
||||
isStreaming.value ||
|
||||
isCancelling.value ||
|
||||
inputBlockedBySuspension.value ||
|
||||
(!isStreaming.value && hasInFlightToolCalls.value),
|
||||
!hasDraft.value &&
|
||||
!isLoadingHistory.value &&
|
||||
(isStreaming.value ||
|
||||
isCancelling.value ||
|
||||
hasOpenInteraction.value ||
|
||||
hasOpenSuspension.value ||
|
||||
hasInFlightToolCalls.value),
|
||||
);
|
||||
|
||||
const chatPlaceholder = computed(() => {
|
||||
if (hasOpenApproval.value) {
|
||||
return locale.baseText('agents.chat.approval.inputPlaceholder');
|
||||
}
|
||||
if (inputBlockedBySuspension.value) {
|
||||
return locale.baseText('agents.chat.waiting.inputPlaceholder');
|
||||
}
|
||||
if (hasOpenInteractiveQuestion.value) {
|
||||
return locale.baseText('agents.chat.answerQuestionPlaceholder');
|
||||
}
|
||||
|
|
@ -460,6 +505,14 @@ watch(
|
|||
},
|
||||
);
|
||||
|
||||
watch(
|
||||
() => [props.projectId, props.agentId, props.continueSessionId],
|
||||
() => {
|
||||
queuedExternalMessage = undefined;
|
||||
queueEdit.value = undefined;
|
||||
},
|
||||
);
|
||||
|
||||
function consumeQueuedExternalMessage(message: string) {
|
||||
if (queuedExternalMessage !== message) return;
|
||||
queuedExternalMessage = undefined;
|
||||
|
|
@ -489,7 +542,6 @@ async function onSubmit(): Promise<SubmitResult> {
|
|||
if (inputText.value.trim() === text) inputText.value = '';
|
||||
consumeQueuedExternalMessage(text);
|
||||
});
|
||||
if (isCurrentTarget()) consumeQueuedExternalMessage(text);
|
||||
return result === 'busy' ? 'busy' : 'sent';
|
||||
}
|
||||
|
||||
|
|
@ -507,24 +559,20 @@ async function onSubmit(): Promise<SubmitResult> {
|
|||
props.connectedTriggers,
|
||||
);
|
||||
if (!isCurrentTarget()) return 'rejected';
|
||||
// Keep the draft if a local resume or cancellation started during preparation.
|
||||
if (isStreaming.value || isCancelling.value) return 'busy';
|
||||
|
||||
agentTelemetry.trackSubmittedMessage({
|
||||
agentId: props.agentId,
|
||||
status: props.agentStatus,
|
||||
agentConfig: fingerprint,
|
||||
});
|
||||
|
||||
const sending = sendMessage(text, files.length > 0 ? files : undefined, () => {
|
||||
if (!isCurrentTarget()) return;
|
||||
agentTelemetry.trackSubmittedMessage({
|
||||
agentId: props.agentId,
|
||||
status: props.agentStatus,
|
||||
agentConfig: fingerprint,
|
||||
});
|
||||
if (inputText.value.trim() === text) inputText.value = '';
|
||||
attachedFiles.value = attachedFiles.value.filter((file) => !files.includes(file));
|
||||
consumeQueuedExternalMessage(text);
|
||||
});
|
||||
isPreparingToSend.value = false;
|
||||
const result = await sending;
|
||||
if (isCurrentTarget()) consumeQueuedExternalMessage(text);
|
||||
if (result === 'busy') return 'busy';
|
||||
return 'sent';
|
||||
} finally {
|
||||
|
|
@ -533,7 +581,6 @@ async function onSubmit(): Promise<SubmitResult> {
|
|||
}
|
||||
|
||||
function sendMessageFromOutside(message: string) {
|
||||
if (inputBlockedBySuspension.value) return;
|
||||
queuedExternalMessage = message;
|
||||
inputText.value = message;
|
||||
void submitQueuedExternalMessage();
|
||||
|
|
@ -555,7 +602,7 @@ async function submitQueuedExternalMessage() {
|
|||
if (result === 'rejected' && queuedExternalMessage === message) {
|
||||
queuedExternalMessage = undefined;
|
||||
}
|
||||
if (queuedExternalMessage && !isSubmissionBlocked.value) {
|
||||
if (queuedExternalMessage && queuedExternalMessage !== message && !isSubmissionBlocked.value) {
|
||||
await nextTick();
|
||||
void submitQueuedExternalMessage();
|
||||
}
|
||||
|
|
@ -595,14 +642,17 @@ onBeforeUnmount(() => {
|
|||
</span>
|
||||
</div>
|
||||
<template #trailingContent>
|
||||
<N8nIconButton
|
||||
icon="x"
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:aria-label="locale.baseText('agents.chat.misconfigured.dismiss')"
|
||||
:title="locale.baseText('agents.chat.misconfigured.dismiss')"
|
||||
@click="dismissFatalError"
|
||||
/>
|
||||
<N8nTooltip :content="locale.baseText('agents.chat.misconfigured.dismiss')" placement="top">
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:aria-label="locale.baseText('agents.chat.misconfigured.dismiss')"
|
||||
@click="dismissFatalError"
|
||||
>
|
||||
<template #icon><N8nIcon icon="x" size="xsmall" aria-hidden="true" /></template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
</template>
|
||||
</N8nCallout>
|
||||
|
||||
|
|
@ -625,14 +675,17 @@ onBeforeUnmount(() => {
|
|||
}}</span>
|
||||
</div>
|
||||
<template #trailingContent>
|
||||
<N8nIconButton
|
||||
icon="x"
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:aria-label="locale.baseText('agents.chat.warning.dismiss')"
|
||||
:title="locale.baseText('agents.chat.warning.dismiss')"
|
||||
@click="dismissWarning(index)"
|
||||
/>
|
||||
<N8nTooltip :content="locale.baseText('agents.chat.warning.dismiss')" placement="top">
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:aria-label="locale.baseText('agents.chat.warning.dismiss')"
|
||||
@click="dismissWarning(index)"
|
||||
>
|
||||
<template #icon><N8nIcon icon="x" size="xsmall" aria-hidden="true" /></template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
</template>
|
||||
</N8nCallout>
|
||||
</div>
|
||||
|
|
@ -651,93 +704,216 @@ onBeforeUnmount(() => {
|
|||
/>
|
||||
|
||||
<div :class="$style.inputArea">
|
||||
<div
|
||||
v-if="showBackgroundJobs"
|
||||
ref="backgroundJobCard"
|
||||
:class="$style.backgroundJobs"
|
||||
data-testid="agent-background-jobs"
|
||||
>
|
||||
<N8nAiActivityStepGroup
|
||||
:key="continueSessionId"
|
||||
:label="backgroundTitle"
|
||||
full-width
|
||||
content-position="above"
|
||||
>
|
||||
<template #prefix>
|
||||
<N8nIcon
|
||||
:icon="backgroundRunningCount ? 'loader-circle' : 'circle'"
|
||||
:spin="backgroundRunningCount > 0"
|
||||
size="small"
|
||||
:class="{ [$style.jobSpinner]: backgroundRunningCount > 0 }"
|
||||
aria-hidden="true"
|
||||
/>
|
||||
</template>
|
||||
<template #header-trailing>
|
||||
<span
|
||||
:class="$style.jobTimer"
|
||||
aria-live="off"
|
||||
data-testid="agent-background-jobs-timer"
|
||||
>{{ backgroundElapsed }}</span
|
||||
>
|
||||
</template>
|
||||
<div :class="$style.backgroundJobDetails">
|
||||
<ul :class="$style.backgroundJobList">
|
||||
<li v-for="job in backgroundJobRows" :key="job.id">
|
||||
<span
|
||||
role="img"
|
||||
:aria-label="job.indicator.label"
|
||||
:title="job.indicator.label"
|
||||
:class="[
|
||||
$style.jobStatus,
|
||||
{ [$style.jobWaiting]: job.indicator.icon === 'circle' },
|
||||
]"
|
||||
:data-status="job.status"
|
||||
>
|
||||
<N8nIcon
|
||||
:icon="job.indicator.icon"
|
||||
:spin="job.indicator.icon === 'loader-circle'"
|
||||
size="small"
|
||||
:class="{ [$style.jobSpinner]: job.indicator.icon === 'loader-circle' }"
|
||||
/>
|
||||
</span>
|
||||
<span>{{ job.label }}</span>
|
||||
</li>
|
||||
</ul>
|
||||
<N8nLink
|
||||
v-if="continueSessionId"
|
||||
:to="backgroundTraceRoute"
|
||||
theme="text"
|
||||
size="small"
|
||||
underline
|
||||
data-testid="agent-background-jobs-trace"
|
||||
>
|
||||
<span :class="$style.jobTraceLabel">
|
||||
<N8nIcon icon="arrow-right" size="small" aria-hidden="true" />
|
||||
{{ locale.baseText('agents.chat.backgroundTasks.viewTrace') }}
|
||||
</span>
|
||||
</N8nLink>
|
||||
</div>
|
||||
</N8nAiActivityStepGroup>
|
||||
</div>
|
||||
<ChatInputBase
|
||||
ref="chatInput"
|
||||
v-model="inputText"
|
||||
:placeholder="chatPlaceholder"
|
||||
:is-streaming="showStopAsPrimaryAction"
|
||||
:is-streaming="false"
|
||||
:show-stop-button="showStop"
|
||||
show-voice
|
||||
:show-attach="showAttach"
|
||||
:accepted-mime-types="acceptedMimeTypes"
|
||||
:can-submit="
|
||||
!inputBlockedBySuspension &&
|
||||
!isStreaming &&
|
||||
!isCancelling &&
|
||||
!isPreparingToSend &&
|
||||
(inputText.trim().length > 0 || attachedFiles.length > 0)
|
||||
"
|
||||
:disabled="inputBlockedBySuspension || isPreparingToSend"
|
||||
:can-submit="!isSubmissionBlocked && hasDraft"
|
||||
:disabled="isPreparingToSend"
|
||||
data-testid="chat-input"
|
||||
@submit="onSubmit"
|
||||
@stop="stopGenerating"
|
||||
@files-selected="handleFilesSelected"
|
||||
>
|
||||
<template v-if="showBackgroundJobs" #header>
|
||||
<div
|
||||
ref="backgroundJobCard"
|
||||
:class="$style.backgroundJobs"
|
||||
data-testid="agent-background-jobs"
|
||||
>
|
||||
<template v-if="queueRows.length" #header>
|
||||
<div :class="$style.messageQueue" data-testid="agent-message-queue">
|
||||
<DefineQueueList v-slot="{ items }">
|
||||
<ul :class="[$style.backgroundJobList, $style.queueList]">
|
||||
<li v-for="item in items" :key="item.id" data-testid="agent-queued-message">
|
||||
<div :class="$style.queuePreview" :title="item.message">
|
||||
<N8nInput
|
||||
v-if="queueEdit?.item.id === item.id"
|
||||
v-model="queueEdit.text"
|
||||
type="textarea"
|
||||
size="small"
|
||||
:autosize="{ minRows: 1, maxRows: 6 }"
|
||||
:readonly="queueEdit.unavailable || queueEdit.saving"
|
||||
:aria-label="locale.baseText('agents.chat.queue.edit')"
|
||||
autofocus
|
||||
@keydown="onQueueEditKeydown"
|
||||
/>
|
||||
<span v-else-if="item.message">{{ item.message }}</span>
|
||||
<p
|
||||
v-if="queueEdit?.item.id === item.id && queueEdit.unavailable"
|
||||
:class="$style.queueEditNotice"
|
||||
role="status"
|
||||
>
|
||||
{{ locale.baseText('agents.chat.queue.editUnavailable') }}
|
||||
</p>
|
||||
<span v-for="attachment in item.attachments" :key="attachment.id">{{
|
||||
attachment.fileName
|
||||
}}</span>
|
||||
</div>
|
||||
<div :class="$style.queueActions">
|
||||
<template v-if="queueEdit?.item.id === item.id">
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.save')"
|
||||
:disabled="!canSaveQueueEdit"
|
||||
placement="top"
|
||||
>
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:disabled="!canSaveQueueEdit"
|
||||
:aria-label="locale.baseText('agents.chat.queue.save')"
|
||||
@click="saveQueueEdit"
|
||||
>
|
||||
<template #icon
|
||||
><N8nIcon icon="check" size="large" aria-hidden="true"
|
||||
/></template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.cancelEdit')"
|
||||
:disabled="queueEdit.saving"
|
||||
placement="top"
|
||||
>
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:disabled="queueEdit.saving"
|
||||
:aria-label="locale.baseText('agents.chat.queue.cancelEdit')"
|
||||
@click="queueEdit = undefined"
|
||||
>
|
||||
<template #icon
|
||||
><N8nIcon icon="x" size="large" aria-hidden="true"
|
||||
/></template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
</template>
|
||||
<template v-else>
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.edit')"
|
||||
:disabled="!!queueEdit || removingQueueIds.has(item.id)"
|
||||
placement="top"
|
||||
>
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:disabled="!!queueEdit || removingQueueIds.has(item.id)"
|
||||
:aria-label="locale.baseText('agents.chat.queue.edit')"
|
||||
@click="startQueueEdit(item)"
|
||||
>
|
||||
<template #icon
|
||||
><N8nIcon icon="pencil" size="large" aria-hidden="true"
|
||||
/></template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
<N8nTooltip
|
||||
:content="locale.baseText('agents.chat.queue.remove')"
|
||||
:disabled="removingQueueIds.has(item.id)"
|
||||
placement="top"
|
||||
>
|
||||
<N8nButton
|
||||
icon-only
|
||||
variant="ghost"
|
||||
size="xsmall"
|
||||
:disabled="removingQueueIds.has(item.id)"
|
||||
:aria-label="locale.baseText('agents.chat.queue.remove')"
|
||||
@click="removeQueuedMessage(item.id)"
|
||||
>
|
||||
<template #icon>
|
||||
<N8nIcon icon="trash-2" size="large" aria-hidden="true" />
|
||||
</template>
|
||||
</N8nButton>
|
||||
</N8nTooltip>
|
||||
</template>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
</DefineQueueList>
|
||||
<QueueList :items="queueRows.slice(0, 2)" />
|
||||
<N8nAiActivityStepGroup
|
||||
v-if="queueRows.length > 2"
|
||||
:key="continueSessionId"
|
||||
:label="backgroundTitle"
|
||||
:label="
|
||||
queuedMessages.length > 2
|
||||
? locale.baseText('agents.chat.queue.title', {
|
||||
adjustToNumber: queuedMessages.length - 2,
|
||||
interpolate: { count: queuedMessages.length - 2 },
|
||||
})
|
||||
: locale.baseText('agents.chat.queue.edit')
|
||||
"
|
||||
full-width
|
||||
content-position="above"
|
||||
>
|
||||
<template #prefix>
|
||||
<N8nIcon
|
||||
:icon="backgroundRunningCount ? 'loader-circle' : 'circle'"
|
||||
:spin="backgroundRunningCount > 0"
|
||||
size="small"
|
||||
:class="{ [$style.jobSpinner]: backgroundRunningCount > 0 }"
|
||||
aria-hidden="true"
|
||||
/>
|
||||
</template>
|
||||
<template #header-trailing>
|
||||
<span
|
||||
:class="$style.jobTimer"
|
||||
aria-live="off"
|
||||
data-testid="agent-background-jobs-timer"
|
||||
>{{ backgroundElapsed }}</span
|
||||
>
|
||||
</template>
|
||||
<div :class="$style.backgroundJobDetails">
|
||||
<ul :class="$style.backgroundJobList">
|
||||
<li v-for="job in backgroundJobRows" :key="job.id">
|
||||
<span
|
||||
role="img"
|
||||
:aria-label="job.indicator.label"
|
||||
:title="job.indicator.label"
|
||||
:class="[
|
||||
$style.jobStatus,
|
||||
{ [$style.jobWaiting]: job.indicator.icon === 'circle' },
|
||||
]"
|
||||
:data-status="job.status"
|
||||
>
|
||||
<N8nIcon
|
||||
:icon="job.indicator.icon"
|
||||
:spin="job.indicator.icon === 'loader-circle'"
|
||||
size="small"
|
||||
:class="{ [$style.jobSpinner]: job.indicator.icon === 'loader-circle' }"
|
||||
/>
|
||||
</span>
|
||||
<span>{{ job.label }}</span>
|
||||
</li>
|
||||
</ul>
|
||||
<N8nLink
|
||||
v-if="continueSessionId"
|
||||
:to="backgroundTraceRoute"
|
||||
theme="text"
|
||||
size="small"
|
||||
underline
|
||||
data-testid="agent-background-jobs-trace"
|
||||
>
|
||||
<span :class="$style.jobTraceLabel">
|
||||
<N8nIcon icon="arrow-right" size="small" aria-hidden="true" />
|
||||
{{ locale.baseText('agents.chat.backgroundTasks.viewTrace') }}
|
||||
</span>
|
||||
</N8nLink>
|
||||
</div>
|
||||
<QueueList :items="queueRows.slice(2)" />
|
||||
</N8nAiActivityStepGroup>
|
||||
</div>
|
||||
</template>
|
||||
|
|
@ -754,12 +930,6 @@ onBeforeUnmount(() => {
|
|||
</template>
|
||||
<template #footer-start>
|
||||
<slot name="footer-start" />
|
||||
<N8nSendStopButton
|
||||
v-if="showSuspensionStopAlongsideSend"
|
||||
streaming
|
||||
stop-button-test-id="agent-chat-suspended-stop-button"
|
||||
@stop="stopGenerating"
|
||||
/>
|
||||
</template>
|
||||
</ChatInputBase>
|
||||
</div>
|
||||
|
|
@ -798,9 +968,8 @@ onBeforeUnmount(() => {
|
|||
margin: 0 auto;
|
||||
}
|
||||
|
||||
.backgroundJobs {
|
||||
margin: calc(-1 * var(--spacing--2xs)) calc(-1 * var(--spacing--2xs)) 0;
|
||||
border-bottom: var(--border);
|
||||
.backgroundJobs,
|
||||
.messageQueue {
|
||||
min-width: 0;
|
||||
|
||||
--ai-activity-step--height: auto;
|
||||
|
|
@ -809,6 +978,25 @@ onBeforeUnmount(() => {
|
|||
--ai-activity-step--color: var(--text-color);
|
||||
}
|
||||
|
||||
.backgroundJobs {
|
||||
background: var(--background--surface);
|
||||
box-shadow: var(--shadow--outline), var(--shadow--xs);
|
||||
border-radius: var(--radius--xs);
|
||||
}
|
||||
|
||||
.messageQueue {
|
||||
--text-color: var(--text-color--subtle);
|
||||
|
||||
margin: calc(-1 * var(--spacing--2xs)) calc(-1 * var(--spacing--2xs)) 0;
|
||||
background: var(--background--subtle);
|
||||
border-radius: var(--radius--lg) var(--radius--lg) 0 0;
|
||||
border-bottom: var(--border);
|
||||
}
|
||||
|
||||
.messageQueue :global(.n8n-icon) {
|
||||
color: light-dark(var(--color--neutral-600), var(--color--neutral-400));
|
||||
}
|
||||
|
||||
.backgroundJobDetails {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
|
|
@ -839,6 +1027,55 @@ onBeforeUnmount(() => {
|
|||
}
|
||||
}
|
||||
|
||||
.messageQueue :global(button[aria-expanded]),
|
||||
.queueList > li {
|
||||
font-size: var(--font-size--2xs);
|
||||
}
|
||||
|
||||
.queueList > li {
|
||||
align-items: center;
|
||||
padding-inline: var(--spacing--sm);
|
||||
color: var(--text-color);
|
||||
border-bottom: var(--border);
|
||||
line-height: var(--line-height--md);
|
||||
}
|
||||
|
||||
.messageQueue > .queueList {
|
||||
padding-block: var(--spacing--2xs);
|
||||
|
||||
&:last-child > li:last-child {
|
||||
border-bottom: 0;
|
||||
}
|
||||
}
|
||||
|
||||
.messageQueue > .queueList:not(:last-child) {
|
||||
padding-bottom: 0;
|
||||
}
|
||||
|
||||
.queueActions {
|
||||
display: flex;
|
||||
align-self: center;
|
||||
flex-shrink: 0;
|
||||
}
|
||||
|
||||
.queueEditNotice {
|
||||
margin: var(--spacing--3xs) 0;
|
||||
font-size: var(--font-size--2xs);
|
||||
}
|
||||
|
||||
.queuePreview {
|
||||
flex: 1;
|
||||
min-width: 0;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
|
||||
span {
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
}
|
||||
|
||||
.jobStatus {
|
||||
display: inline-flex;
|
||||
flex-shrink: 0;
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ import type {
|
|||
AgentBackgroundJobsResponse,
|
||||
AgentCapabilitySummary,
|
||||
AgentChatMessagesResponse,
|
||||
AgentChatQueueResponse,
|
||||
AgentChatQueueUpdateDto,
|
||||
AgentConfigMutationResponse,
|
||||
AgentConfigResponse,
|
||||
AgentConfigValidationResponse,
|
||||
|
|
@ -545,6 +547,49 @@ export const getAgentBackgroundJobs = async (
|
|||
);
|
||||
};
|
||||
|
||||
export const getAgentChatQueue = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
): Promise<AgentChatQueueResponse> => {
|
||||
return await makeRestApiRequest(
|
||||
context,
|
||||
'GET',
|
||||
`/projects/${encodeURIComponent(projectId)}/agents/v2/${encodeURIComponent(agentId)}/chat/${encodeURIComponent(threadId)}/queue`,
|
||||
);
|
||||
};
|
||||
|
||||
export const updateAgentQueuedMessage = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
queueId: string,
|
||||
payload: AgentChatQueueUpdateDto,
|
||||
): Promise<void> => {
|
||||
await makeRestApiRequest(
|
||||
context,
|
||||
'PATCH',
|
||||
`/projects/${encodeURIComponent(projectId)}/agents/v2/${encodeURIComponent(agentId)}/chat/${encodeURIComponent(threadId)}/queue/${encodeURIComponent(queueId)}`,
|
||||
payload,
|
||||
);
|
||||
};
|
||||
|
||||
export const removeAgentQueuedMessage = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
queueId: string,
|
||||
): Promise<{ removed: boolean }> => {
|
||||
return await makeRestApiRequest(
|
||||
context,
|
||||
'DELETE',
|
||||
`/projects/${encodeURIComponent(projectId)}/agents/v2/${encodeURIComponent(agentId)}/chat/${encodeURIComponent(threadId)}/queue/${encodeURIComponent(queueId)}`,
|
||||
);
|
||||
};
|
||||
|
||||
export const getChatMessages = async (
|
||||
context: IRestApiContext,
|
||||
projectId: string,
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { useRootStore } from '@n8n/stores/useRootStore';
|
|||
import { isRecord } from '@n8n/utils/is-record';
|
||||
import type {
|
||||
AgentBuilderOpenSuspension,
|
||||
AgentChatQueueItem,
|
||||
AgentPersistedMessageDto,
|
||||
AgentSseEvent,
|
||||
CancellationResumeData,
|
||||
|
|
@ -19,6 +20,9 @@ import {
|
|||
cancelAgentChatRun,
|
||||
clearTestChatMessages,
|
||||
getChatMessages,
|
||||
getAgentChatQueue,
|
||||
removeAgentQueuedMessage,
|
||||
updateAgentQueuedMessage,
|
||||
getTestChatMessages,
|
||||
} from './useAgentApi';
|
||||
|
||||
|
|
@ -81,6 +85,7 @@ type ResumePayload =
|
|||
};
|
||||
|
||||
const STOP_ACCEPTANCE_TIMEOUT_MS = 30 * TIME.SECOND;
|
||||
const MAX_WAITING_STREAMS = 2;
|
||||
|
||||
function getApprovalDecision(value: unknown): boolean | undefined {
|
||||
if (!isRecord(value) || typeof value.approved !== 'boolean') return undefined;
|
||||
|
|
@ -98,6 +103,12 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
|
||||
const messages = ref<ChatMessage[]>([]);
|
||||
const isStreamOpen = ref(false);
|
||||
const isSubmitting = ref(false);
|
||||
const queuedMessages = ref<AgentChatQueueItem[]>([]);
|
||||
const removingQueueIds = ref(new Set<string>());
|
||||
const streams = new Map<AbortController, StreamSession>();
|
||||
let queueVersion = 0;
|
||||
let submissionVersion = 0;
|
||||
const activeExecutionId = ref<string | null>(null);
|
||||
const acceptedSessionId = ref<string>();
|
||||
const isRecovering = ref(false);
|
||||
|
|
@ -109,6 +120,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
const abortController = ref<AbortController | null>(null);
|
||||
const streamSettlements = new WeakMap<AbortController, Promise<void>>();
|
||||
const historyLoaded = ref(false);
|
||||
const isLoadingHistory = ref(false);
|
||||
const pushStore = usePushConnectionStore();
|
||||
const visibility = useDocumentVisibility();
|
||||
let disposed = false;
|
||||
|
|
@ -135,14 +147,14 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
}
|
||||
/**
|
||||
* Set when the backend rejects the stream because the agent itself is
|
||||
* misconfigured (missing instructions / model / credential). Cleared on the
|
||||
* next send so users can fix the config and retry without a manual dismiss.
|
||||
* misconfigured (missing instructions / model / credential). Cleared when the
|
||||
* next execution starts so users can retry without a manual dismiss.
|
||||
*/
|
||||
const fatalError = ref<FatalAgentError | null>(null);
|
||||
/**
|
||||
* Non-fatal warnings emitted during a run (e.g. an MCP server that failed to
|
||||
* connect, so its tools were skipped). The run continues; these are shown to
|
||||
* the user as a warning callout. Visible warnings clear on the next send;
|
||||
* the user as a warning callout. Visible warnings clear when the next execution starts;
|
||||
* explicitly dismissed warnings stay hidden for this composable instance.
|
||||
*/
|
||||
const warnings = ref<AgentChatWarning[]>([]);
|
||||
|
|
@ -239,9 +251,13 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
|
||||
async function loadHistory(): Promise<void> {
|
||||
if (historyLoaded.value) return;
|
||||
isLoadingHistory.value = true;
|
||||
isRecovering.value = true;
|
||||
const queue = refreshQueue();
|
||||
const loaded = await refreshHistory({ clearOnNotFound: true });
|
||||
await queue;
|
||||
historyLoaded.value = true;
|
||||
isLoadingHistory.value = false;
|
||||
// A running resume can have no messages yet. Keep its session selected.
|
||||
if (loaded && (messages.value.length > 0 || !isStreaming.value)) {
|
||||
params.onHistoryLoaded?.(messages.value.length);
|
||||
|
|
@ -259,6 +275,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
...(params.continueSessionId ? { threadId: params.continueSessionId } : {}),
|
||||
},
|
||||
async () => {
|
||||
await refreshQueue();
|
||||
// Defer history refreshes until the local stream ends to preserve streamed text.
|
||||
if (isStreamOpen.value) {
|
||||
refreshAfterStream = true;
|
||||
|
|
@ -273,6 +290,102 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
},
|
||||
);
|
||||
|
||||
async function refreshQueue(): Promise<void> {
|
||||
const threadId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
if (!threadId || disposed) return;
|
||||
const target = targetKey();
|
||||
const version = ++queueVersion;
|
||||
try {
|
||||
const result = await getAgentChatQueue(
|
||||
rootStore.restApiContext,
|
||||
params.projectId.value,
|
||||
params.agentId.value,
|
||||
threadId,
|
||||
);
|
||||
if (!disposed && target === targetKey() && version === queueVersion)
|
||||
queuedMessages.value = result.items;
|
||||
} catch (error) {
|
||||
if (!disposed && target === targetKey() && version === queueVersion) {
|
||||
showError(error, locale.baseText('agents.chat.queue.loadError'));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function updateQueuedMessage(
|
||||
queueId: string,
|
||||
message: string,
|
||||
): Promise<'updated' | 'unavailable' | 'failed'> {
|
||||
const threadId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
if (!threadId || disposed) return 'failed';
|
||||
const target = targetKey();
|
||||
try {
|
||||
await updateAgentQueuedMessage(
|
||||
rootStore.restApiContext,
|
||||
params.projectId.value,
|
||||
params.agentId.value,
|
||||
threadId,
|
||||
queueId,
|
||||
{ message },
|
||||
);
|
||||
if (disposed || target !== targetKey()) return 'failed';
|
||||
queueVersion++;
|
||||
queuedMessages.value = queuedMessages.value.map((item) =>
|
||||
item.id === queueId ? { ...item, message: message.trim() } : item,
|
||||
);
|
||||
return 'updated';
|
||||
} catch (error) {
|
||||
if (disposed || target !== targetKey()) return 'failed';
|
||||
const status = isRecord(error) ? error.httpStatusCode : undefined;
|
||||
if (status === 404 || status === 409) return 'unavailable';
|
||||
showError(error, locale.baseText('agents.chat.queue.editError'));
|
||||
return 'failed';
|
||||
} finally {
|
||||
if (!disposed && target === targetKey()) refreshHistoryFromPush();
|
||||
}
|
||||
}
|
||||
|
||||
async function removeQueuedMessage(queueId: string): Promise<void> {
|
||||
const threadId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
if (!threadId || removingQueueIds.value.has(queueId)) return;
|
||||
const target = targetKey();
|
||||
removingQueueIds.value.add(queueId);
|
||||
try {
|
||||
await removeAgentQueuedMessage(
|
||||
rootStore.restApiContext,
|
||||
params.projectId.value,
|
||||
params.agentId.value,
|
||||
threadId,
|
||||
queueId,
|
||||
);
|
||||
if (disposed || target !== targetKey()) return;
|
||||
queueVersion++;
|
||||
queuedMessages.value = queuedMessages.value.filter((item) => item.id !== queueId);
|
||||
for (const [controller, session] of streams) {
|
||||
if (session.queueId === queueId && !session.executionId) controller.abort();
|
||||
}
|
||||
} catch (error) {
|
||||
if (!disposed && target === targetKey())
|
||||
showError(error, locale.baseText('agents.chat.queue.removeError'));
|
||||
} finally {
|
||||
if (target === targetKey()) {
|
||||
removingQueueIds.value.delete(queueId);
|
||||
refreshHistoryFromPush();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const removeQueueListener = pushStore.addEventListener((event) => {
|
||||
if (
|
||||
event.type === 'agentMessageQueueUpdated' &&
|
||||
event.data.projectId === params.projectId.value &&
|
||||
event.data.agentId === params.agentId.value &&
|
||||
event.data.threadId === (params.continueSessionId?.value ?? acceptedSessionId.value)
|
||||
) {
|
||||
queueVersion++;
|
||||
refreshHistoryFromPush();
|
||||
}
|
||||
});
|
||||
|
||||
// Recover missed updates when the preview reopens, reconnects, becomes visible, or changes session.
|
||||
function refresh() {
|
||||
retryCount = 0;
|
||||
|
|
@ -291,6 +404,9 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
watch(targetKey, () => {
|
||||
detachStream();
|
||||
messages.value = [];
|
||||
queuedMessages.value = [];
|
||||
removingQueueIds.value.clear();
|
||||
queueVersion++;
|
||||
activeExecutionId.value = null;
|
||||
acceptedSessionId.value = undefined;
|
||||
isRecovering.value = true;
|
||||
|
|
@ -303,6 +419,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
// Clear retry timers and ignore late responses when this chat closes.
|
||||
onScopeDispose(() => {
|
||||
disposed = true;
|
||||
removeQueueListener();
|
||||
detachStream();
|
||||
clearTimeout(retryTimer);
|
||||
});
|
||||
|
|
@ -326,6 +443,9 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
|
||||
interface StreamSession {
|
||||
target: string;
|
||||
controller: AbortController;
|
||||
queueId?: string;
|
||||
userMessage?: ChatMessage;
|
||||
executionId?: string;
|
||||
busy?: boolean;
|
||||
onAccepted?: () => void;
|
||||
|
|
@ -538,11 +658,53 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
event: AgentSseEvent,
|
||||
session: StreamSession,
|
||||
): { done?: boolean } | undefined {
|
||||
if (session.controller.signal.aborted) return { done: true };
|
||||
if (session.executionId && abortController.value !== session.controller) {
|
||||
finalizeStream(session);
|
||||
return { done: true };
|
||||
}
|
||||
// Controller validation failures only emit `error`. Any other event proves
|
||||
// that the backend admitted and persisted this turn.
|
||||
if (event.type !== 'error') acknowledgeSessionCreation();
|
||||
switch (event.type) {
|
||||
case 'message-queued':
|
||||
session.queueId = event.queueId;
|
||||
acceptedSessionId.value = event.sessionId;
|
||||
detachExcessWaitingStreams();
|
||||
session.onAccepted?.();
|
||||
session.onAccepted = undefined;
|
||||
queueVersion++;
|
||||
refreshHistoryFromPush();
|
||||
break;
|
||||
case 'execution-started':
|
||||
if (session.userMessage) {
|
||||
// Local messages enter FIFO order. An earlier request cannot own a later turn.
|
||||
for (const [controller, earlier] of streams) {
|
||||
if (controller === session.controller) break;
|
||||
if (earlier.userMessage) controller.abort();
|
||||
}
|
||||
}
|
||||
abortController.value = session.controller;
|
||||
isStreamOpen.value = true;
|
||||
streamVersion++;
|
||||
fatalError.value = null;
|
||||
warnings.value = [];
|
||||
if (session.userMessage) {
|
||||
if (
|
||||
!messages.value.some(
|
||||
(message) => message.role === 'user' && message.executionId === event.executionId,
|
||||
)
|
||||
) {
|
||||
messages.value.push({
|
||||
...session.userMessage,
|
||||
content: event.message ?? session.userMessage.content,
|
||||
executionId: event.executionId,
|
||||
});
|
||||
}
|
||||
}
|
||||
queueVersion++;
|
||||
queuedMessages.value = queuedMessages.value.filter((item) => item.id !== session.queueId);
|
||||
void refreshQueue();
|
||||
clearTimeout(stopAcceptanceTimer);
|
||||
session.executionId = event.executionId;
|
||||
activeExecutionId.value = event.executionId;
|
||||
|
|
@ -766,6 +928,8 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
markInFlightStateFailed(session);
|
||||
if (event.errorCode === 'agent_misconfigured') {
|
||||
fatalError.value = { message: event.message, missing: event.missing ?? [] };
|
||||
} else if (session.userMessage && !session.executionId) {
|
||||
showError(new Error(event.message), locale.baseText('agents.chat.queue.sendError'));
|
||||
} else {
|
||||
messages.value.push(
|
||||
reactive<ChatMessage>({
|
||||
|
|
@ -796,6 +960,15 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
return undefined;
|
||||
}
|
||||
|
||||
function detachExcessWaitingStreams(): void {
|
||||
let waiting = 0;
|
||||
for (const [controller, session] of streams) {
|
||||
if (!session.queueId || session.executionId || controller.signal.aborted) continue;
|
||||
// Leave HTTP/1.1 connections available for controls and history recovery.
|
||||
if (++waiting > MAX_WAITING_STREAMS) controller.abort();
|
||||
}
|
||||
}
|
||||
|
||||
async function consumeStream(
|
||||
response: Response,
|
||||
session: StreamSession,
|
||||
|
|
@ -857,8 +1030,12 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
url: string,
|
||||
body: Record<string, unknown>,
|
||||
onAccepted?: () => void,
|
||||
userMessage?: ChatMessage,
|
||||
): Promise<{ outcome: StreamOutcome }> {
|
||||
const controller = new AbortController();
|
||||
const session: StreamSession = {
|
||||
controller,
|
||||
userMessage,
|
||||
target: targetKey(),
|
||||
onAccepted,
|
||||
errorEmitted: false,
|
||||
|
|
@ -867,10 +1044,12 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
reasoningStartedAt: new Map(),
|
||||
openReasoning: new Map(),
|
||||
};
|
||||
isStreamOpen.value = true;
|
||||
streamVersion++;
|
||||
const controller = new AbortController();
|
||||
abortController.value = controller;
|
||||
streams.set(controller, session);
|
||||
if (!userMessage) {
|
||||
isStreamOpen.value = true;
|
||||
streamVersion++;
|
||||
abortController.value = controller;
|
||||
}
|
||||
let settleStream: (() => void) | undefined;
|
||||
streamSettlements.set(
|
||||
controller,
|
||||
|
|
@ -891,32 +1070,40 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
});
|
||||
if (!isCurrent() || controller.signal.aborted) return { outcome: 'aborted' };
|
||||
if (!response.ok || !response.body) {
|
||||
messages.value.push({
|
||||
id: crypto.randomUUID(),
|
||||
role: 'assistant',
|
||||
content: `Error: ${response.statusText || 'Failed to reach agent'}`,
|
||||
status: 'error',
|
||||
});
|
||||
handleEvent(
|
||||
{ type: 'error', message: response.statusText || 'Failed to reach agent' },
|
||||
session,
|
||||
);
|
||||
return { outcome: 'failed' };
|
||||
}
|
||||
await consumeStream(response, session, controller.signal);
|
||||
if (!isCurrent() || controller.signal.aborted) return { outcome: 'aborted' };
|
||||
if (session.busy) return { outcome: 'busy' };
|
||||
if (!session.terminalEventReceived) {
|
||||
isRecovering.value = true;
|
||||
if (session.userMessage && !session.queueId && !session.executionId)
|
||||
showError(
|
||||
new Error('Stream closed before acceptance'),
|
||||
locale.baseText('agents.chat.queue.sendError'),
|
||||
);
|
||||
if (abortController.value === controller) isRecovering.value = true;
|
||||
return { outcome: 'detached' };
|
||||
}
|
||||
finalizeStream(session);
|
||||
if (!session.errorEmitted) session.onAccepted?.();
|
||||
return { outcome: session.errorEmitted ? 'failed' : 'completed' };
|
||||
} catch {
|
||||
} catch (error) {
|
||||
if (!isCurrent() || controller.signal.aborted) return { outcome: 'aborted' };
|
||||
if (session.errorEmitted) return { outcome: 'failed' };
|
||||
if (session.userMessage && !session.queueId && !session.executionId)
|
||||
showError(error, locale.baseText('agents.chat.queue.sendError'));
|
||||
// A lost response cannot tell us whether the server accepted or finished the turn.
|
||||
isRecovering.value = true;
|
||||
if (abortController.value === controller) isRecovering.value = true;
|
||||
return { outcome: 'detached' };
|
||||
} finally {
|
||||
streams.delete(controller);
|
||||
if (abortController.value === controller) {
|
||||
if (session.terminalEventReceived && session.executionId === activeExecutionId.value)
|
||||
activeExecutionId.value = null;
|
||||
clearTimeout(stopAcceptanceTimer);
|
||||
abortController.value = null;
|
||||
isStreamOpen.value = false;
|
||||
|
|
@ -924,8 +1111,13 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
streamSettlements.delete(controller);
|
||||
settleStream?.();
|
||||
if (isCurrent()) {
|
||||
streamVersion++;
|
||||
if (session.executionId || refreshAfterStream || isRecovering.value) {
|
||||
if (session.executionId) streamVersion++;
|
||||
if (
|
||||
session.userMessage ||
|
||||
session.executionId ||
|
||||
refreshAfterStream ||
|
||||
isRecovering.value
|
||||
) {
|
||||
refreshAfterStream = false;
|
||||
refreshHistoryFromPush();
|
||||
} else if (isCancelling.value && !activeExecutionId.value) {
|
||||
|
|
@ -935,13 +1127,18 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
}
|
||||
}
|
||||
|
||||
async function streamChat(message: string, files?: File[], onAccepted?: () => void) {
|
||||
async function streamChat(
|
||||
message: string,
|
||||
files?: File[],
|
||||
onAccepted?: () => void,
|
||||
userMessage?: ChatMessage,
|
||||
) {
|
||||
const target = targetKey();
|
||||
const { baseUrl } = rootStore.restApiContext;
|
||||
const url = `${baseUrl}/projects/${params.projectId.value}/agents/v2/${params.agentId.value}/chat`;
|
||||
const body: Record<string, unknown> = { message };
|
||||
const sessionId = params.continueSessionId?.value;
|
||||
const newSession = params.newSession?.value === true;
|
||||
const sessionId = params.continueSessionId?.value ?? acceptedSessionId.value;
|
||||
const newSession = params.newSession?.value === true && sessionId !== acknowledgedSessionId;
|
||||
if (sessionId) {
|
||||
body.sessionId = sessionId;
|
||||
if (newSession) body.newSession = true;
|
||||
|
|
@ -961,7 +1158,7 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
);
|
||||
}
|
||||
if (disposed || target !== targetKey()) return { outcome: 'aborted' };
|
||||
const result = await postAndConsume(url, body, onAccepted);
|
||||
const result = await postAndConsume(url, body, onAccepted, userMessage);
|
||||
if (
|
||||
newSession &&
|
||||
params.newSession?.value === true &&
|
||||
|
|
@ -1120,38 +1317,49 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
onAccepted?: () => void,
|
||||
): Promise<'sent' | 'busy'> {
|
||||
const trimmed = text.trim();
|
||||
if ((!trimmed && !files?.length) || isStreaming.value || isCancelling.value) return 'busy';
|
||||
isStreamOpen.value = true;
|
||||
const messageId = crypto.randomUUID();
|
||||
if ((!trimmed && !files?.length) || isSubmitting.value || isLoadingHistory.value) return 'busy';
|
||||
isSubmitting.value = true;
|
||||
const submission = ++submissionVersion;
|
||||
const target = targetKey();
|
||||
// Any new send invalidates a prior misconfig banner — the user is retrying.
|
||||
fatalError.value = null;
|
||||
warnings.value = [];
|
||||
messages.value.push({
|
||||
id: messageId,
|
||||
const userMessage: ChatMessage = {
|
||||
id: crypto.randomUUID(),
|
||||
role: 'user',
|
||||
content: trimmed,
|
||||
status: 'success',
|
||||
createdAt: Date.now(),
|
||||
...(files?.length && {
|
||||
attachments: files.map((file) => ({
|
||||
fileName: file.name,
|
||||
mimeType: resolveFileMimeType(file.name, file.type) || 'application/octet-stream',
|
||||
sizeBytes: file.size,
|
||||
file,
|
||||
})),
|
||||
}),
|
||||
attachments: files?.map((file) => ({
|
||||
fileName: file.name,
|
||||
mimeType: resolveFileMimeType(file.name, file.type) || 'application/octet-stream',
|
||||
sizeBytes: file.size,
|
||||
file,
|
||||
})),
|
||||
};
|
||||
return await new Promise<'sent' | 'busy'>((resolve) => {
|
||||
let released = false;
|
||||
const release = (outcome: 'sent' | 'busy' = 'sent') => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
if (submission === submissionVersion) isSubmitting.value = false;
|
||||
resolve(outcome);
|
||||
};
|
||||
void streamChat(
|
||||
trimmed,
|
||||
files,
|
||||
() => {
|
||||
onAccepted?.();
|
||||
release();
|
||||
},
|
||||
userMessage,
|
||||
)
|
||||
.then(({ outcome }) => {
|
||||
release(outcome === 'busy' ? 'busy' : 'sent');
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
if (!disposed && target === targetKey())
|
||||
showError(error, locale.baseText('agents.chat.queue.sendError'));
|
||||
})
|
||||
.finally(release);
|
||||
});
|
||||
try {
|
||||
const { outcome } = await streamChat(trimmed, files, onAccepted);
|
||||
if (outcome === 'busy') {
|
||||
messages.value = messages.value.filter((message) => message.id !== messageId);
|
||||
return 'busy';
|
||||
}
|
||||
return 'sent';
|
||||
} finally {
|
||||
if (!abortController.value && target === targetKey()) isStreamOpen.value = false;
|
||||
}
|
||||
}
|
||||
|
||||
function dismissFatalError(): void {
|
||||
|
|
@ -1168,10 +1376,11 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
|
||||
function detachStream(): void {
|
||||
clearTimeout(stopAcceptanceTimer);
|
||||
const controller = abortController.value;
|
||||
abortController.value = null;
|
||||
isStreamOpen.value = false;
|
||||
controller?.abort();
|
||||
isSubmitting.value = false;
|
||||
submissionVersion++;
|
||||
for (const controller of streams.keys()) controller.abort();
|
||||
}
|
||||
|
||||
function retainStopUntilAcceptance(): void {
|
||||
|
|
@ -1257,13 +1466,19 @@ export function useAgentChatStream(params: UseAgentChatStreamParams) {
|
|||
await refreshHistory();
|
||||
showError(error, locale.baseText('agents.chat.stop.error'));
|
||||
} finally {
|
||||
detachStream();
|
||||
controller?.abort();
|
||||
await settlement;
|
||||
isCancelling.value = false;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
queuedMessages,
|
||||
removingQueueIds,
|
||||
removeQueuedMessage,
|
||||
updateQueuedMessage,
|
||||
isSubmitting,
|
||||
isLoadingHistory,
|
||||
messages,
|
||||
isStreaming,
|
||||
activeExecutionId,
|
||||
|
|
|
|||
|
|
@ -84,13 +84,31 @@ describe('ChatInputBase', () => {
|
|||
expect(queryByTestId('instance-ai-stop-button')).not.toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('should show stop button when streaming', () => {
|
||||
const { getByTestId, queryByTestId } = renderComponent({
|
||||
props: makeProps({ isStreaming: true }),
|
||||
it('switches Stop independently of composer tools when requested', async () => {
|
||||
const { getByTestId, queryByTestId, rerender, emitted } = renderComponent({
|
||||
props: makeProps({
|
||||
isStreaming: true,
|
||||
canSubmit: false,
|
||||
showAttach: true,
|
||||
showVoice: true,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(getByTestId('instance-ai-stop-button')).toBeInTheDocument();
|
||||
expect(queryByTestId('instance-ai-send-button')).not.toBeInTheDocument();
|
||||
expect(getByTestId('chat-input-attach-button')).toBeDisabled();
|
||||
expect(getByTestId('chat-input-voice-button')).toBeDisabled();
|
||||
|
||||
await rerender({ isStreaming: false, showStopButton: true });
|
||||
expect(getByTestId('instance-ai-stop-button')).toBeEnabled();
|
||||
expect(queryByTestId('instance-ai-send-button')).not.toBeInTheDocument();
|
||||
expect(getByTestId('chat-input-attach-button')).toBeEnabled();
|
||||
expect(getByTestId('chat-input-voice-button')).toBeEnabled();
|
||||
|
||||
await rerender({ showStopButton: false, canSubmit: true, modelValue: 'Follow-up' });
|
||||
expect(queryByTestId('instance-ai-stop-button')).not.toBeInTheDocument();
|
||||
getByTestId('instance-ai-send-button').click();
|
||||
expect(emitted().submit).toEqual([[]]);
|
||||
});
|
||||
|
||||
it('should emit submit on Enter keydown', () => {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ const props = withDefaults(
|
|||
modelValue: string;
|
||||
placeholder?: string;
|
||||
isStreaming: boolean;
|
||||
/** Override the action button separately from the composer busy state. */
|
||||
showStopButton?: boolean;
|
||||
canSubmit: boolean;
|
||||
disabled?: boolean;
|
||||
showVoice?: boolean;
|
||||
|
|
@ -43,6 +45,7 @@ const props = withDefaults(
|
|||
}>(),
|
||||
{
|
||||
placeholder: undefined,
|
||||
showStopButton: undefined,
|
||||
acceptedMimeTypes: undefined,
|
||||
attachedEncodedBytes: 0,
|
||||
autosize: () => ({ minRows: 2, maxRows: 6 }),
|
||||
|
|
@ -262,7 +265,7 @@ defineExpose({
|
|||
ref="inputRef"
|
||||
:model-value="modelValue"
|
||||
:placeholder="placeholder"
|
||||
:streaming="isStreaming"
|
||||
:streaming="showStopButton ?? isStreaming"
|
||||
:disabled="disabled"
|
||||
:submit-disabled="!canSubmit"
|
||||
:button-label="props.buttonLabel"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import type {
|
||||
AgentChatMessageDto,
|
||||
AgentChatMessagesResponse,
|
||||
AgentChatQueueResponse,
|
||||
AgentJsonConfig,
|
||||
AgentSseEvent,
|
||||
} from '@n8n/api-types';
|
||||
|
|
@ -64,6 +65,32 @@ export class AgentApiHelper {
|
|||
return (await response.json()).data;
|
||||
}
|
||||
|
||||
async queuedMessages(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
): Promise<AgentChatQueueResponse> {
|
||||
const response = await this.api.request.get(
|
||||
`/rest/projects/${projectId}/agents/v2/${agentId}/chat/${threadId}/queue`,
|
||||
);
|
||||
if (!response.ok())
|
||||
throw new TestError(`Failed to read queued messages: ${await response.text()}`);
|
||||
return (await response.json()).data;
|
||||
}
|
||||
|
||||
async removeQueuedMessage(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
threadId: string,
|
||||
queueId: string,
|
||||
): Promise<void> {
|
||||
const response = await this.api.request.delete(
|
||||
`/rest/projects/${projectId}/agents/v2/${agentId}/chat/${threadId}/queue/${queueId}`,
|
||||
);
|
||||
if (!response.ok())
|
||||
throw new TestError(`Failed to remove queued message: ${await response.text()}`);
|
||||
}
|
||||
|
||||
async executions(
|
||||
projectId: string,
|
||||
agentId: string,
|
||||
|
|
|
|||
|
|
@ -171,6 +171,29 @@ test.describe(
|
|||
const third = await open(0, threadId, 'third');
|
||||
await expect.poll(async () => await queued(threadId)).toBe(2);
|
||||
third.disconnect();
|
||||
// A fresh reader restores accepted input from another main without resubmission.
|
||||
const removable = await open(0, threadId, 'remove before processing');
|
||||
await expect
|
||||
.poll(() => removable.events.find((event) => event.type === 'message-queued'))
|
||||
.toBeTruthy();
|
||||
const restored = await consumer.agents.queuedMessages(project.id, agent.id, threadId);
|
||||
expect(restored.items.map(({ message }) => message)).toEqual([
|
||||
'fifo-remote',
|
||||
'third',
|
||||
'remove before processing',
|
||||
]);
|
||||
await consumer.agents.removeQueuedMessage(
|
||||
project.id,
|
||||
agent.id,
|
||||
threadId,
|
||||
restored.items[2].id,
|
||||
);
|
||||
expect(
|
||||
(await ingress.agents.queuedMessages(project.id, agent.id, threadId)).items.map(
|
||||
({ message }) => message,
|
||||
),
|
||||
).toEqual(['fifo-remote', 'third']);
|
||||
removable.disconnect();
|
||||
const independent = await open(1, randomUUID(), 'independent', true);
|
||||
expect(await independent.done).toBeUndefined();
|
||||
expect(independent.events).toContainEqual(expect.objectContaining({ type: 'done' }));
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user