fix(web): address review findings for Nexus AI stop/cancel

- Fix race conditions in useAppState.tsx abort lifecycle:
  - Replace stale isChatLoading closure guard with chatStateRef
  - Track and cancel rAF handles in stopChatResponse/finally
  - Move cancelled chunk check before onChunk dispatch
  - Simplify finally block to unconditional cleanup via chatStateRef
  - Guard tool_result from overwriting stopped status
  - Have clearChat abort in-flight streams before clearing
- Reorder isAbortError to check error identity before signal.aborted
- Refactor AgentStreamChunk to discriminated union for exhaustive switch
- Fix test assertions to use exact .toEqual() per DoD §2.7
- Add test for plain Error with name AbortError
- Remove dead markStopped alias, simplify signal spread-conditional
This commit is contained in:
Test 2026-05-26 18:27:32 +01:00
parent fced6a8a55
commit 3ffd54dab1
4 changed files with 65 additions and 43 deletions

View file

@ -368,9 +368,9 @@ export interface AgentRuntimeOptions {
}
const isAbortError = (error: unknown, signal?: AbortSignal): boolean => {
if (signal?.aborted) return true;
if (error instanceof DOMException && error.name === 'AbortError') return true;
if (error instanceof Error && error.name === 'AbortError') return true;
if (signal?.aborted) return true;
return false;
};
@ -447,7 +447,7 @@ export async function* streamAgentResponse(
streamMode: ['values', 'messages'] as any,
// Allow longer tool/reasoning loops (more Cursor-like persistence)
recursionLimit: 50,
...(options.signal ? { signal: options.signal } : {}),
signal: options.signal,
} as any);
// Track what we've yielded to avoid duplicates
@ -525,10 +525,11 @@ export async function* streamAgentResponse(
// - After all tools are done: treat as final content
const isReasoning =
!hasSeenToolCallThisTurn || toolCalls.length > 0 || pendingToolCalls > 0;
yield {
type: isReasoning ? 'reasoning' : 'content',
[isReasoning ? 'reasoning' : 'content']: content,
};
if (isReasoning) {
yield { type: 'reasoning', reasoning: content };
} else {
yield { type: 'content', content };
}
}
// Track tool calls from message chunks

View file

@ -286,22 +286,17 @@ export type AgentHistoryMessage =
};
/**
* Streaming chunk from agent
* Now supports step-based streaming where each step is a distinct message
* Streaming chunk from agent (discriminated union).
* Each variant carries only its relevant fields, enabling exhaustive switch handling.
*/
export interface AgentStreamChunk {
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) */
content?: string;
/** Hidden raw transcript for reconstructing future agent turns */
historyMessages?: AgentHistoryMessage[];
/** Tool call information */
toolCall?: ToolCallInfo;
/** Error message */
error?: string;
}
export type AgentStreamChunk =
| { type: 'reasoning'; reasoning: string }
| { type: 'tool_call'; toolCall: ToolCallInfo }
| { type: 'tool_result'; toolCall: ToolCallInfo }
| { type: 'content'; content: string }
| { type: 'error'; error: string }
| { type: 'done'; historyMessages?: AgentHistoryMessage[] }
| { type: 'cancelled' };
/**
* A single step in the agent's execution

View file

@ -589,6 +589,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 chatStateRef = useRef<'idle' | 'streaming' | 'aborting'>('idle');
const initializeAgent = useCallback(
async (overrideProjectName?: string): Promise<void> => {
@ -647,7 +648,7 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
const sendChatMessage = useCallback(
async (message: string): Promise<void> => {
if (isChatLoading) return;
if (chatStateRef.current !== 'idle') return;
// Refresh Code panel for the new question: keep user-pinned refs, clear old AI citations
clearAICodeReferences();
@ -686,6 +687,7 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
}
setIsChatLoading(true);
chatStateRef.current = 'streaming';
setCurrentToolCalls([]);
chatAbortRef.current?.abort();
@ -747,11 +749,13 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
});
};
let pendingUpdate = false;
let rafHandle: number | null = null;
const scheduleMessageUpdate = () => {
if (pendingUpdate) return;
pendingUpdate = true;
requestAnimationFrame(() => {
rafHandle = requestAnimationFrame(() => {
pendingUpdate = false;
rafHandle = null;
updateMessage();
});
};
@ -898,7 +902,7 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
if (idx < 0) {
idx = toolCallsForMessage.findIndex((t) => t.name === tc.name && !t.result);
}
if (idx >= 0) {
if (idx >= 0 && toolCallsForMessage[idx].status !== 'stopped') {
toolCallsForMessage[idx] = {
...toolCallsForMessage[idx],
result: tc.result,
@ -914,7 +918,11 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
(s.toolCall.id === tc.id ||
(s.toolCall.name === tc.name && s.toolCall.status === 'running')),
);
if (stepIdx >= 0 && stepsForMessage[stepIdx].toolCall) {
if (
stepIdx >= 0 &&
stepsForMessage[stepIdx].toolCall &&
stepsForMessage[stepIdx].toolCall!.status !== 'stopped'
) {
stepsForMessage[stepIdx] = {
...stepsForMessage[stepIdx],
toolCall: {
@ -935,6 +943,8 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
targetIdx = prev.findIndex((t) => t.name === tc.name && !t.result);
}
if (targetIdx >= 0) {
const target = prev[targetIdx];
if (target.status === 'stopped') return prev;
return prev.map((t, i) =>
i === targetIdx ? { ...t, result: tc.result, status: 'completed' } : t,
);
@ -1035,10 +1045,10 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
captureHistory: providerCapabilities.preserveAssistantTranscript,
signal: chatAbortController.signal,
})) {
onChunk(chunk);
if (chunk.type === 'cancelled') {
break;
}
onChunk(chunk);
}
} catch (error) {
if (!chatAbortController.signal.aborted) {
@ -1046,15 +1056,13 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
setAgentError(message);
}
} finally {
if (chatAbortRef.current === chatAbortController) {
chatAbortRef.current = null;
setIsChatLoading(false);
setCurrentToolCalls([]);
} else if (!chatAbortRef.current) {
// Stopped with no new turn started — safe to clear loading UI
setIsChatLoading(false);
setCurrentToolCalls([]);
if (rafHandle != null) {
cancelAnimationFrame(rafHandle);
rafHandle = null;
}
chatStateRef.current = 'idle';
setIsChatLoading(false);
setCurrentToolCalls([]);
}
},
[
@ -1068,14 +1076,14 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
clearAIToolHighlights,
graph,
embeddingStatus,
isChatLoading,
],
);
const stopChatResponse = useCallback(() => {
if (!isChatLoading) return;
if (!chatAbortRef.current) return;
chatAbortRef.current?.abort();
chatStateRef.current = 'aborting';
chatAbortRef.current.abort();
chatAbortRef.current = null;
const stoppedLabel = i18n.t('chat:stopped');
@ -1094,14 +1102,13 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
if (lastAssistantIdx === undefined) return prev;
const message = prev[lastAssistantIdx];
const markStopped = markStoppedToolCall;
const updated: ChatMessage = {
...message,
toolCalls: message.toolCalls?.map(markStopped),
toolCalls: message.toolCalls?.map(markStoppedToolCall),
steps: message.steps?.map((step) =>
step.type === 'tool_call' && step.toolCall
? { ...step, toolCall: markStopped(step.toolCall) }
? { ...step, toolCall: markStoppedToolCall(step.toolCall) }
: step,
),
};
@ -1109,12 +1116,16 @@ const AppStateProviderInner = ({ children }: { children: ReactNode }) => {
});
setIsChatLoading(false);
}, [isChatLoading]);
}, []);
const clearChat = useCallback(() => {
chatAbortRef.current?.abort();
chatAbortRef.current = null;
chatStateRef.current = 'idle';
setChatMessages([]);
setCurrentToolCalls([]);
setAgentError(null);
setIsChatLoading(false);
}, []);
// Switch to a different repo on the connected server

View file

@ -41,8 +41,8 @@ describe('streamAgentResponse abort', () => {
if (chunk.type === 'cancelled') break;
}
expect(chunks.some((c) => c.type === 'cancelled')).toBe(true);
expect(chunks.some((c) => c.type === 'error')).toBe(false);
expect(chunks[chunks.length - 1]).toEqual({ type: 'cancelled' });
expect(chunks.filter((c) => c.type === 'error')).toEqual([]);
});
it('passes AbortSignal to agent.stream config', async () => {
@ -65,6 +65,21 @@ describe('streamAgentResponse abort', () => {
expect(capturedConfig?.signal).toBe(controller.signal);
});
it('yields cancelled for a plain Error with name AbortError', async () => {
const agent = {
stream: async () => {
throw Object.assign(new Error('aborted'), { name: 'AbortError' });
},
};
const chunks = [];
for await (const chunk of streamAgentResponse(agent as any, userMessage)) {
chunks.push(chunk);
}
expect(chunks).toEqual([{ type: 'cancelled' }]);
});
it('does not treat unrelated errors mentioning abort as cancellation', async () => {
const agent = {
stream: async () => {