Commit bdb82411 authored by edy's avatar edy

feat(chat): support cancelling active streams

parent 5abfbb86
Pipeline #18469 failed
This diff is collapsed.
......@@ -95,6 +95,7 @@ const desktopApi: DesktopApi = {
readImageAttachmentDataUrl: (attachment: ChatAttachment) => ipcRenderer.invoke(IPC_CHANNELS.chatReadImageAttachmentDataUrl, attachment),
sendPrompt: (sessionId: string, prompt: string, skillId?: string, attachments?: ChatAttachment[]) => ipcRenderer.invoke(IPC_CHANNELS.chatSendPrompt, sessionId, prompt, skillId, attachments),
streamPrompt: (sessionId: string, prompt: string, skillId?: string, attachments?: ChatAttachment[]) => ipcRenderer.invoke(IPC_CHANNELS.chatStreamPrompt, sessionId, prompt, skillId, attachments),
cancelStream: (requestId: string, runId?: string, sessionId?: string) => ipcRenderer.invoke(IPC_CHANNELS.chatCancelStream, requestId, runId, sessionId),
onStreamEvent: (listener: ChatStreamListener) => {
const wrapped = (_event: Electron.IpcRendererEvent, payload: Parameters<ChatStreamListener>[0]) => {
listener(payload);
......
import test from "node:test"
import assert from "node:assert/strict"
import { readFileSync } from "node:fs"
const ipcSource = readFileSync(new URL("../src/main/ipc.ts", import.meta.url), "utf8")
const preloadSource = readFileSync(new URL("../src/preload/index.ts", import.meta.url), "utf8")
test("desktop IPC registers chat cancel stream handler", () => {
assert.match(ipcSource, /activeChatStreams/)
assert.match(ipcSource, /cancelStream\s*=\s*async/)
assert.match(ipcSource, /IPC_CHANNELS\.chatCancelStream/)
assert.match(preloadSource, /cancelStream: \(requestId: string, runId\?: string, sessionId\?: string\)/)
})
test("desktop cancel marks assistant stream as stopped and broadcasts cancelled event", () => {
assert.match(ipcSource, /statusLabel:\s*"已停止"/)
assert.match(ipcSource, /type:\s*"cancelled"/)
assert.match(ipcSource, /gatewayClient\.cancelChatRun/)
})
......@@ -31,6 +31,7 @@ import {
NavIcon,
RedBookIcon,
RefreshIcon,
StopIcon,
ThumbIcon,
TrashIcon,
getIntentSuggestionIcon,
......@@ -703,7 +704,8 @@ export default function App() {
sending,
streamSmoke,
activeStreamRef,
submitPrompt
submitPrompt,
cancelActiveStream
} = useChatStreamingController({
desktopApi,
viewMode,
......@@ -784,9 +786,9 @@ export default function App() {
normalizeError: err
});
const sendButtonLabel = sendPhase === "preparing"
? ui.preparing
? "停止生成"
: sendPhase === "streaming" || sendPhase === "finalizing"
? ui.generating
? "停止生成"
: !isBound
? ui.bindFirst
: ui.send;
......@@ -1429,8 +1431,9 @@ export default function App() {
skills: effectiveSkills,
skillMenuOpen,
attachmentIcon: <AttachmentIcon />,
submitIcon: <ArrowUpIcon />,
submitIcon: sendPhase !== "idle" ? <StopIcon /> : <ArrowUpIcon />,
onSubmit: sendPrompt,
onCancel: cancelActiveStream,
onPromptChange: setPrompt,
onTextareaKeyDown: handleComposerKeyDown,
onAttachmentSelection: handleAttachmentSelection,
......
......@@ -328,6 +328,14 @@ export function ArrowUpIcon() {
);
}
export function StopIcon() {
return (
<svg viewBox="0 0 24 24" fill="none" aria-hidden="true" focusable="false">
<rect x="7" y="7" width="10" height="10" rx="1.8" fill="currentColor" />
</svg>
);
}
export function RefreshIcon() {
return (
<svg viewBox="0 0 24 24" fill="none" aria-hidden="true" focusable="false">
......
......@@ -40,6 +40,7 @@ interface ChatComposerProps {
attachmentIcon: ReactNode
submitIcon: ReactNode
onSubmit: () => void | Promise<void>
onCancel: () => void | Promise<void>
onPromptChange: (value: string) => void
onTextareaKeyDown: (event: ReactKeyboardEvent<HTMLTextAreaElement>) => void | Promise<void>
onAttachmentSelection: (event: ChangeEvent<HTMLInputElement>) => void
......@@ -74,6 +75,7 @@ export function ChatComposer({
attachmentIcon,
submitIcon,
onSubmit,
onCancel,
onPromptChange,
onTextareaKeyDown,
onAttachmentSelection,
......@@ -159,13 +161,14 @@ export function ChatComposer({
</button>
</div>
<button
type="submit"
type={sending ? "button" : "submit"}
className={"composer-submit" + (sending ? " is-busy" : "")}
disabled={!canSend}
disabled={sending ? false : !canSend}
onClick={sending ? onCancel : undefined}
aria-label={sendButtonLabel}
title={sendButtonLabel}
>
{sending ? <span className="composer-submit-spinner" aria-hidden="true" /> : submitIcon}
{submitIcon}
<span className="visually-hidden">{sendButtonLabel}</span>
</button>
</div>
......
......@@ -130,6 +130,7 @@ interface ConversationWorkspaceViewProps {
attachmentIcon: ReactNode
submitIcon: ReactNode
onSubmit: () => void | Promise<void>
onCancel: () => void | Promise<void>
onPromptChange: (value: string) => void
onTextareaKeyDown: (event: ReactKeyboardEvent<HTMLTextAreaElement>) => void | Promise<void>
onAttachmentSelection: (event: ChangeEvent<HTMLInputElement>) => void
......@@ -221,6 +222,7 @@ export function ConversationWorkspaceView({
attachmentIcon,
submitIcon,
onSubmit,
onCancel,
onPromptChange,
onTextareaKeyDown,
onAttachmentSelection,
......@@ -387,6 +389,7 @@ export function ConversationWorkspaceView({
attachmentIcon={attachmentIcon}
submitIcon={submitIcon}
onSubmit={onSubmit}
onCancel={onCancel}
onPromptChange={onPromptChange}
onTextareaKeyDown={onTextareaKeyDown}
onAttachmentSelection={onAttachmentSelection}
......
export type SmokeStreamPhase = "idle" | "requested" | "started" | "streaming" | "completed" | "fallback" | "error"
export type SmokeStreamPhase = "idle" | "requested" | "started" | "streaming" | "completed" | "fallback" | "error" | "cancelled"
export interface SmokeStreamSnapshot {
phase: SmokeStreamPhase
......
......@@ -24,6 +24,7 @@ const mockUi = {
waitingReply: "已收到问题,正在组织回答"
} as const
const mockChatStreamListeners = new Set<ChatStreamListener>();
const mockChatStreamTimers = new Map<string, number[]>();
function emitMockChatStreamEvent(event: ChatStreamEvent) {
for (const listener of mockChatStreamListeners) {
......@@ -401,21 +402,28 @@ export const mockDesktopApi = {
const executionPolicy = { source: "client-config" as const, modelId: "qwen3.6-plus", modelLabel: "qwen3.6-plus", routingMode: "platform-managed" as const, skillId, skillName: skillId, message: "mock" };
const replyText = "Mock: " + prompt;
const chunks = replyText.match(/.{1,6}/g) ?? [replyText];
const timers: number[] = [];
const scheduleStreamTimer = (handler: () => void, delay: number) => {
const timer = window.setTimeout(handler, delay);
timers.push(timer);
};
mockChatStreamTimers.set(requestId, timers);
let fullText = "";
window.setTimeout(() => {
scheduleStreamTimer(() => {
emitMockChatStreamEvent({ type: "status", requestId, sessionId, runId, stage: "prepare-request", label: mockUi.preparingReply });
emitMockChatStreamEvent({ type: "started", requestId, sessionId, runId, executionPolicy });
}, 0);
window.setTimeout(() => {
scheduleStreamTimer(() => {
emitMockChatStreamEvent({ type: "status", requestId, sessionId, runId, stage: "await-model", label: mockUi.waitingReply });
}, 30);
chunks.forEach((chunk, index) => {
window.setTimeout(() => {
scheduleStreamTimer(() => {
fullText += chunk;
emitMockChatStreamEvent({ type: "delta", requestId, sessionId, runId, textDelta: chunk, fullText });
}, 90 * (index + 1));
});
window.setTimeout(() => {
scheduleStreamTimer(() => {
mockChatStreamTimers.delete(requestId);
emitMockChatStreamEvent({
type: "completed",
requestId,
......@@ -427,6 +435,28 @@ export const mockDesktopApi = {
}, 90 * (chunks.length + 1));
return { requestId, sessionId, runId, userMessageId, assistantMessageId, executionPolicy };
},
cancelStream: async (requestId: string, runId?: string, sessionId?: string) => {
for (const timer of mockChatStreamTimers.get(requestId) ?? []) {
window.clearTimeout(timer);
}
mockChatStreamTimers.delete(requestId);
emitMockChatStreamEvent({
type: "cancelled",
requestId,
sessionId,
runId,
message: "已停止",
remoteCancelled: false
});
return {
requestId,
sessionId,
runId,
localCancelled: true,
remoteCancelled: false,
message: "已停止"
};
},
onStreamEvent: (listener: ChatStreamListener) => {
mockChatStreamListeners.add(listener);
return () => {
......
import test from "node:test"
import assert from "node:assert/strict"
import { readFileSync } from "node:fs"
const controllerSource = readFileSync(new URL("../src/features/chat/useChatStreamingController.ts", import.meta.url), "utf8")
const composerSource = readFileSync(new URL("../src/features/chat/ChatComposer.tsx", import.meta.url), "utf8")
const mockSource = readFileSync(new URL("../src/lib/mock-desktop-api.ts", import.meta.url), "utf8")
test("chat streaming controller exposes cancelActiveStream and ignores later events", () => {
assert.match(controllerSource, /cancelActiveStream/)
assert.match(controllerSource, /desktopApi\.chat\.cancelStream/)
assert.match(controllerSource, /stoppedRequestIdsRef/)
assert.match(controllerSource, /event\.type === "cancelled"/)
})
test("chat streaming controller scopes preparing cancellation per submission", () => {
assert.match(controllerSource, /cancelledSubmissionIdsRef/)
assert.match(controllerSource, /pendingSubmissionIdRef/)
assert.doesNotMatch(controllerSource, /preStreamCancelRequestedRef/)
})
test("chat streaming controller clears stopped request ids on terminal events", () => {
assert.match(controllerSource, /stoppedRequestIdsRef\.current\.delete\(event\.requestId\)/)
})
test("composer can submit a stop action while sending", () => {
assert.match(composerSource, /onCancel/)
assert.match(composerSource, /type=\{sending \? "button" : "submit"\}/)
assert.match(composerSource, /onClick=\{sending \? onCancel : undefined\}/)
})
test("mock desktop API cancels pending stream timers", () => {
assert.match(mockSource, /mockChatStreamTimers/)
assert.match(mockSource, /cancelStream/)
assert.match(mockSource, /window\.clearTimeout/)
})
......@@ -114,6 +114,12 @@ export interface GatewayPromptStreamStart {
completion: Promise<ChatMessage>;
}
export interface GatewayCancelChatRunResult {
runId: string;
localCancelled: boolean;
remoteCancelled: boolean;
}
export interface GatewayPromptStreamDelta {
sessionId: string;
runId: string;
......@@ -483,6 +489,45 @@ export class GatewayClient {
return { sessionId, runId, completion };
}
async cancelChatRun(runId: string): Promise<GatewayCancelChatRunResult> {
const pending = this.pendingChatRuns.get(runId);
if (pending) {
clearTimeout(pending.timer);
this.pendingChatRuns.delete(runId);
pending.resolve({
id: `${pending.sessionKey}:${runId}:cancelled`,
role: "assistant",
content: pending.accumulatedText,
createdAt: new Date().toISOString()
});
}
const availableMethods = this.statusSnapshot.availableMethods ?? [];
if (!availableMethods.includes("chat.cancel")) {
return {
runId,
localCancelled: Boolean(pending),
remoteCancelled: false
};
}
try {
await this.request("chat.cancel", { runId });
return {
runId,
localCancelled: Boolean(pending),
remoteCancelled: true
};
} catch (error) {
this.appendLog("warn", `Gateway chat.cancel failed for ${runId}: ${error instanceof Error ? error.message : String(error)}`);
return {
runId,
localCancelled: Boolean(pending),
remoteCancelled: false
};
}
}
private async handleEvent(frame: Record<string, unknown>): Promise<void> {
const eventName = String(frame.event ?? "unknown");
......@@ -1187,4 +1232,3 @@ export class GatewayClient {
return message;
}
}
import test from "node:test"
import assert from "node:assert/strict"
import { readFileSync } from "node:fs"
const gatewaySource = readFileSync(new URL("../src/index.ts", import.meta.url), "utf8")
test("gateway client cancels local pending run even when remote cancel is unavailable", () => {
assert.match(gatewaySource, /async cancelChatRun\(runId: string\)/)
assert.match(gatewaySource, /remoteCancelled: false/)
assert.match(gatewaySource, /this\.pendingChatRuns\.delete\(runId\)/)
})
test("gateway client only sends cancel RPC when gateway advertises chat cancel", () => {
assert.match(gatewaySource, /availableMethods.*chat\.cancel/s)
assert.match(gatewaySource, /this\.request\("chat\.cancel"/)
})
......@@ -38,6 +38,7 @@
chatReadImageAttachmentDataUrl: "chat:read-image-attachment-data-url",
chatSendPrompt: "chat:send-prompt",
chatStreamPrompt: "chat:stream-prompt",
chatCancelStream: "chat:cancel-stream",
chatStreamEvent: "chat:stream-event",
diagnosticsOpenControlUi: "diagnostics:open-control-ui",
diagnosticsExportSnapshot: "diagnostics:export-snapshot",
......@@ -509,6 +510,21 @@ export interface ChatStreamPromptResult {
executionPolicy?: ChatExecutionPolicy;
}
export interface ChatCancelStreamRequest {
requestId: string;
runId?: string;
sessionId?: string;
}
export interface ChatCancelStreamResult {
requestId: string;
sessionId?: string;
runId?: string;
localCancelled: boolean;
remoteCancelled: boolean;
message: string;
}
export interface ChatStreamStartedEvent {
type: "started";
requestId: string;
......@@ -554,7 +570,16 @@ export interface ChatStreamErrorEvent {
errorCategory?: string;
}
export type ChatStreamEvent = ChatStreamStartedEvent | ChatStreamStatusEvent | ChatStreamDeltaEvent | ChatStreamCompletedEvent | ChatStreamErrorEvent;
export interface ChatStreamCancelledEvent {
type: "cancelled";
requestId: string;
sessionId?: string;
runId?: string;
message: string;
remoteCancelled: boolean;
}
export type ChatStreamEvent = ChatStreamStartedEvent | ChatStreamStatusEvent | ChatStreamDeltaEvent | ChatStreamCompletedEvent | ChatStreamErrorEvent | ChatStreamCancelledEvent;
export type ChatStreamListener = (event: ChatStreamEvent) => void;
......@@ -965,6 +990,7 @@ export interface DesktopApi {
readImageAttachmentDataUrl(attachment: ChatAttachment): Promise<string | null>;
sendPrompt(sessionId: string, prompt: string, skillId?: string, attachments?: ChatAttachment[]): Promise<PromptResult>;
streamPrompt(sessionId: string, prompt: string, skillId?: string, attachments?: ChatAttachment[]): Promise<ChatStreamPromptResult>;
cancelStream(requestId: string, runId?: string, sessionId?: string): Promise<ChatCancelStreamResult>;
onStreamEvent(listener: ChatStreamListener): () => void;
};
diagnostics: {
......
import test from "node:test"
import assert from "node:assert/strict"
import { readFileSync } from "node:fs"
const sharedTypesSource = readFileSync(new URL("../src/index.ts", import.meta.url), "utf8")
test("shared desktop chat API exposes cancel stream contract", () => {
assert.match(sharedTypesSource, /chatCancelStream:\s*"chat:cancel-stream"/)
assert.match(sharedTypesSource, /export interface ChatCancelStreamRequest/)
assert.match(sharedTypesSource, /export interface ChatCancelStreamResult/)
assert.match(sharedTypesSource, /cancelStream\(requestId: string, runId\?: string, sessionId\?: string\): Promise<ChatCancelStreamResult>/)
})
test("shared stream events include cancelled payload", () => {
assert.match(sharedTypesSource, /export interface ChatStreamCancelledEvent/)
assert.match(sharedTypesSource, /type:\s*"cancelled"/)
assert.match(sharedTypesSource, /ChatStreamEvent = .*ChatStreamCancelledEvent/s)
})
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment