fix(web): stop Nexus AI agent when user clicks Stop

Wire AbortController through chat streaming so Stop cancels the LangGraph
run instead of only hiding the loading UI. Fixes #1615.
This commit is contained in:
basseyriman 2026-05-25 05:25:11 +01:00
parent 1c4993251c
commit 6918eb662b
6 changed files with 174 additions and 13 deletions

View file

@ -363,8 +363,17 @@ export type AgentMessage = { role: 'user'; content: string } | AgentHistoryMessa
export interface AgentRuntimeOptions {
/** Capture assistant/tool messages for providers that require exact transcript replay. */
captureHistory?: boolean;
/** When aborted (e.g. user clicked Stop), the stream ends with a `cancelled` chunk. */
signal?: AbortSignal;
}
const isAbortError = (error: unknown): boolean => {
if (error instanceof DOMException && error.name === 'AbortError') return true;
if (error instanceof Error && error.name === 'AbortError') return true;
const message = error instanceof Error ? error.message : String(error);
return /aborted|abort/i.test(message);
};
export const buildLangChainMessages = (messages: AgentMessage[]): BaseMessage[] =>
messages.map((message) => {
if (message.role === 'user') {
@ -434,11 +443,15 @@ export async function* streamAgentResponse(
const formattedMessages = buildLangChainMessages(messages);
// Use BOTH modes: 'values' for structure, 'messages' for token streaming
const stream = await agent.stream({ messages: formattedMessages }, {
streamMode: ['values', 'messages'] as any,
// Allow longer tool/reasoning loops (more Cursor-like persistence)
recursionLimit: 50,
} as any);
const stream = await agent.stream(
{ messages: formattedMessages },
{
streamMode: ['values', 'messages'] as any,
// Allow longer tool/reasoning loops (more Cursor-like persistence)
recursionLimit: 50,
...(options.signal ? { signal: options.signal } : {}),
} as any,
);
// Track what we've yielded to avoid duplicates
const yieldedToolCalls = new Set<string>();
@ -455,6 +468,10 @@ export async function* streamAgentResponse(
let lastStepMessages: any[] | null = null;
for await (const event of stream) {
if (options.signal?.aborted) {
break;
}
// Events come as [streamMode, data] tuples when using multiple modes
// or just data when using single mode
let mode: string;
@ -627,6 +644,11 @@ export async function* streamAgentResponse(
}
}
if (options.signal?.aborted) {
yield { type: 'cancelled' };
return;
}
// DEBUG: Stream completed normally
if (import.meta.env.DEV) {
console.log('✅ Stream completed normally, yielding done');
@ -640,6 +662,10 @@ export async function* streamAgentResponse(
: undefined,
};
} catch (error) {
if (options.signal?.aborted || isAbortError(error)) {
yield { type: 'cancelled' };
return;
}
const message = error instanceof Error ? error.message : String(error);
// DEBUG: Stream error
if (import.meta.env.DEV) {

View file

@ -290,7 +290,7 @@ export type AgentHistoryMessage =
* Now supports step-based streaming where each step is a distinct message
*/
export interface AgentStreamChunk {
type: 'reasoning' | 'tool_call' | 'tool_result' | 'content' | 'error' | 'done';
type: 'reasoning' | 'tool_call' | 'tool_result' | 'content' | 'error' | 'done' | 'cancelled';
/** LLM's reasoning/thinking text (shown as a step) */
reasoning?: string;
/** Final answer content (streamed token by token) */

View file

@ -582,6 +582,7 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
// Agent state — agent runs on main thread now (I/O-bound, not CPU-bound)
const agentRef = useRef<any>(null);
const chatAbortRef = useRef<AbortController | null>(null);
const initializeAgent = useCallback(
async (overrideProjectName?: string): Promise<void> => {
@ -681,6 +682,10 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
setIsChatLoading(true);
setCurrentToolCalls([]);
chatAbortRef.current?.abort();
const chatAbortController = new AbortController();
chatAbortRef.current = chatAbortController;
const providerCapabilities = getProviderCapabilities(llmSettings.activeProvider);
// Prepare message history for agent (convert our format to AgentMessage format)
@ -1012,6 +1017,39 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
// Finalize the assistant message - just call updateMessage one more time
scheduleMessageUpdate();
break;
case 'cancelled': {
const stoppedLabel = i18n.t('chat:stopped');
for (let i = 0; i < toolCallsForMessage.length; i++) {
const tc = toolCallsForMessage[i];
if (tc.status === 'running' || tc.status === 'pending') {
toolCallsForMessage[i] = {
...tc,
status: 'error',
result: stoppedLabel,
};
}
}
for (let i = 0; i < stepsForMessage.length; i++) {
const step = stepsForMessage[i];
if (
step.type === 'tool_call' &&
step.toolCall &&
(step.toolCall.status === 'running' || step.toolCall.status === 'pending')
) {
stepsForMessage[i] = {
...step,
toolCall: {
...step.toolCall,
status: 'error',
result: stoppedLabel,
},
};
}
}
scheduleMessageUpdate();
break;
}
}
};
@ -1022,13 +1060,25 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
const { streamAgentResponse } = await import('../core/llm/agent');
for await (const chunk of streamAgentResponse(agent, history, {
captureHistory: providerCapabilities.preserveAssistantTranscript,
signal: chatAbortController.signal,
})) {
if (chatAbortController.signal.aborted) {
break;
}
onChunk(chunk);
if (chunk.type === 'cancelled') {
break;
}
}
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
setAgentError(message);
if (!chatAbortController.signal.aborted) {
const message = error instanceof Error ? error.message : String(error);
setAgentError(message);
}
} finally {
if (chatAbortRef.current === chatAbortController) {
chatAbortRef.current = null;
}
setIsChatLoading(false);
setCurrentToolCalls([]);
}
@ -1049,11 +1099,46 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
);
const stopChatResponse = useCallback(() => {
if (isChatLoading) {
// Agent streaming will be interrupted by the AbortController in sendChatMessage
setIsChatLoading(false);
setCurrentToolCalls([]);
}
if (!isChatLoading) return;
chatAbortRef.current?.abort();
chatAbortRef.current = null;
const stoppedLabel = i18n.t('chat:stopped');
setCurrentToolCalls((prev) =>
prev.map((tc) =>
tc.status === 'running' || tc.status === 'pending'
? { ...tc, status: 'error', result: stoppedLabel }
: tc,
),
);
setChatMessages((prev) => {
const lastAssistantIdx = [...prev]
.map((m, i) => (m.role === 'assistant' ? i : -1))
.filter((i) => i >= 0)
.pop();
if (lastAssistantIdx === undefined) return prev;
const message = prev[lastAssistantIdx];
const markStopped = (tc: ToolCallInfo): ToolCallInfo =>
tc.status === 'running' || tc.status === 'pending'
? { ...tc, status: 'error', result: stoppedLabel }
: tc;
const updated: ChatMessage = {
...message,
toolCalls: message.toolCalls?.map(markStopped),
steps: message.steps?.map((step) =>
step.type === 'tool_call' && step.toolCall
? { ...step, toolCall: markStopped(step.toolCall) }
: step,
),
};
return prev.map((m, i) => (i === lastAssistantIdx ? updated : m));
});
setIsChatLoading(false);
}, [isChatLoading]);
const clearChat = useCallback(() => {

View file

@ -24,6 +24,7 @@
"clearChat": "Clear chat",
"stopResponse": "Stop response"
},
"stopped": "Stopped by user",
"badges": {
"configureAI": "Configure AI",
"connecting": "Connecting"

View file

@ -24,6 +24,7 @@
"clearChat": "清空聊天",
"stopResponse": "停止响应"
},
"stopped": "用户已停止",
"badges": {
"configureAI": "配置 AI",
"connecting": "连接中"

View file

@ -0,0 +1,48 @@
import { describe, expect, it } from 'vitest';
import { streamAgentResponse, type AgentMessage } from '../../src/core/llm/agent';
describe('streamAgentResponse abort', () => {
const userMessage: AgentMessage[] = [{ role: 'user', content: 'hello' }];
it('yields cancelled when the LangGraph stream throws AbortError', async () => {
const agent = {
stream: async () => {
throw new DOMException('The operation was aborted', 'AbortError');
},
};
const chunks = [];
for await (const chunk of streamAgentResponse(agent as any, userMessage, {
signal: new AbortController().signal,
})) {
chunks.push(chunk);
}
expect(chunks).toEqual([{ type: 'cancelled' }]);
});
it('yields cancelled when the abort signal is set mid-stream', async () => {
const controller = new AbortController();
const agent = {
stream: async function* () {
yield ['values', { messages: [] }];
controller.abort();
// Simulate a long-running graph step after abort
for (let i = 0; i < 100; i++) {
yield ['messages', [{ _getType: () => 'ai', content: 'still going' }]];
}
},
};
const chunks = [];
for await (const chunk of streamAgentResponse(agent as any, userMessage, {
signal: controller.signal,
})) {
chunks.push(chunk);
if (chunk.type === 'cancelled') break;
}
expect(chunks.some((c) => c.type === 'cancelled')).toBe(true);
expect(chunks.some((c) => c.type === 'error')).toBe(false);
});
});