diff --git a/web/src/api/chatRetry.test.ts b/web/src/api/chatRetry.test.ts new file mode 100644 index 00000000..694a9261 --- /dev/null +++ b/web/src/api/chatRetry.test.ts @@ -0,0 +1,22 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { shouldAutoRetry, autoRetryDelayMs, fallbackErrorClass } from './chatRetry'; + +test('shouldAutoRetry: transient only, max 2', () => { + assert.equal(shouldAutoRetry({ class: 'transient' }, 0), true); + assert.equal(shouldAutoRetry({ class: 'transient' }, 1), true); + assert.equal(shouldAutoRetry({ class: 'transient' }, 2), false); + assert.equal(shouldAutoRetry({ class: 'fatal' }, 0), false); + assert.equal(shouldAutoRetry({ class: 'resumable' }, 0), false); +}); + +test('autoRetryDelayMs: 1s then 2s', () => { + assert.equal(autoRetryDelayMs(0), 1000); + assert.equal(autoRetryDelayMs(1), 2000); +}); + +test('fallbackErrorClass: network → transient, workspace → resumable', () => { + assert.equal(fallbackErrorClass({ network: true }), 'transient'); + assert.equal(fallbackErrorClass({ workspaceId: 'research_1' }), 'resumable'); + assert.equal(fallbackErrorClass({}), 'fatal'); +}); diff --git a/web/src/api/chatRetry.ts b/web/src/api/chatRetry.ts new file mode 100644 index 00000000..22fcffb6 --- /dev/null +++ b/web/src/api/chatRetry.ts @@ -0,0 +1,21 @@ +import type { ChatErrorInfo } from '../types/workbench'; + +export function shouldAutoRetry( + info: Pick, + attemptIndex: number, +): boolean { + return info.class === 'transient' && attemptIndex < 2; +} + +export function autoRetryDelayMs(attemptIndex: number): number { + return attemptIndex === 0 ? 1000 : 2000; +} + +export function fallbackErrorClass(criteria: { + network?: boolean; + workspaceId?: string; +}): ChatErrorInfo['class'] { + if (criteria.network) return 'transient'; + if (criteria.workspaceId) return 'resumable'; + return 'fatal'; +} diff --git a/web/src/api/chatStream.test.ts b/web/src/api/chatStream.test.ts new file mode 100644 index 00000000..858b9dca --- /dev/null +++ b/web/src/api/chatStream.test.ts @@ -0,0 +1,19 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { parseSSEChunk } from './chatStream'; + +test('parseSSEChunk: parses event and data lines', () => { + const raw = + 'event: phase\ndata: {"type":"phase","phase":{"phase_id":"intent_recognizing","message":"正在研判您的意图与项目上下文..."}}\n\n'; + const { events, rest } = parseSSEChunk(raw); + assert.equal(rest, ''); + assert.equal(events.length, 1); + assert.equal(events[0].type, 'phase'); + assert.equal(events[0].phase?.phase_id, 'intent_recognizing'); +}); + +test('parseSSEChunk: keeps incomplete frame in rest', () => { + const { events, rest } = parseSSEChunk('event: delta\ndata: {"type":"delta"'); + assert.equal(events.length, 0); + assert.ok(rest.includes('event: delta')); +}); diff --git a/web/src/api/chatStream.ts b/web/src/api/chatStream.ts new file mode 100644 index 00000000..d4fa10a3 --- /dev/null +++ b/web/src/api/chatStream.ts @@ -0,0 +1,157 @@ +import type { ChatErrorInfo } from '../types/workbench'; +import { fallbackErrorClass } from './chatRetry'; + +export type StreamEvent = { + id?: string; + type: string; + phase?: { + phase_id?: string; + message?: string; + step_index?: number; + total_steps?: number; + progress?: number; + }; + status?: { message?: string; intent?: string }; + delta?: string; + error?: string; + error_info?: { + user_message?: string; + code?: string; + class?: string; + phase_id?: string; + upstream?: string; + request_id?: string; + session_id?: string; + workspace_id?: string; + actions?: string[]; + }; + done?: { session_id?: string; duration_ms?: number }; +}; + +type ErrorAction = ChatErrorInfo['actions'][number]; + +const VALID_ACTIONS: readonly ErrorAction[] = ['retry', 'continue', 'restart']; + +function isErrorClass(value: unknown): value is ChatErrorInfo['class'] { + return ( + value === 'transient' || + value === 'resumable' || + value === 'fatal' || + value === 'cancelled' + ); +} + +function defaultActionsForClass(cls: ChatErrorInfo['class']): ChatErrorInfo['actions'] { + switch (cls) { + case 'transient': + return ['retry']; + case 'resumable': + return ['continue', 'restart']; + case 'fatal': + return ['restart']; + case 'cancelled': + return ['retry']; + default: { + const _exhaustive: never = cls; + return _exhaustive; + } + } +} + +function parseActions(raw?: string[]): ChatErrorInfo['actions'] { + if (!raw?.length) return []; + return raw.filter((a): a is ErrorAction => (VALID_ACTIONS as readonly string[]).includes(a)); +} + +function parseFrame(frame: string): StreamEvent | null { + if (!frame.trim()) return null; + + let dataLine: string | undefined; + for (const line of frame.split('\n')) { + if (line.startsWith('data:')) { + dataLine = line.slice(5).trim(); + } + } + if (!dataLine) return null; + + return JSON.parse(dataLine) as StreamEvent; +} + +export function parseSSEChunk(buffer: string): { events: StreamEvent[]; rest: string } { + const events: StreamEvent[] = []; + if (!buffer) return { events, rest: '' }; + + const frames = buffer.split('\n\n'); + const hasTrailingDelimiter = buffer.endsWith('\n\n'); + const processCount = hasTrailingDelimiter ? frames.length : Math.max(0, frames.length - 1); + + for (let i = 0; i < processCount; i++) { + const ev = parseFrame(frames[i]); + if (ev) events.push(ev); + } + + const rest = hasTrailingDelimiter ? '' : (frames[frames.length - 1] ?? ''); + return { events, rest }; +} + +export async function streamChat( + body: Record, + handlers: { + onEvent: (ev: StreamEvent) => void; + onRequestId?: (id: string) => void; + }, + signal?: AbortSignal, +): Promise { + const resp = await fetch('/api/v1/chat/stream', { + method: 'POST', + headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' }, + body: JSON.stringify(body), + signal, + }); + const rid = resp.headers.get('X-Request-Id') || resp.headers.get('x-request-id'); + if (rid) handlers.onRequestId?.(rid); + if (!resp.ok || !resp.body) { + throw Object.assign(new Error(`HTTP ${resp.status}`), { + httpStatus: resp.status, + requestId: rid, + }); + } + const reader = resp.body.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + const parsed = parseSSEChunk(buf); + buf = parsed.rest; + for (const ev of parsed.events) handlers.onEvent(ev); + } +} + +export function mapErrorInfo(ev: StreamEvent): ChatErrorInfo { + const ei = ev.error_info ?? {}; + const userMessage = ei.user_message ?? ev.error ?? '发生未知错误'; + const code = ei.code ?? 'UNKNOWN_ERROR'; + + const cls: ChatErrorInfo['class'] = isErrorClass(ei.class) + ? ei.class + : fallbackErrorClass({ workspaceId: ei.workspace_id }); + + let actions = parseActions(ei.actions); + if (!actions.length) { + actions = defaultActionsForClass(cls); + } + + return { + userMessage, + code, + class: cls, + phaseId: ei.phase_id, + upstream: ei.upstream, + requestId: ei.request_id, + sessionId: ei.session_id, + workspaceId: ei.workspace_id, + actions, + }; +} diff --git a/web/src/types/workbench.ts b/web/src/types/workbench.ts index bd133372..bc9f4ec8 100644 --- a/web/src/types/workbench.ts +++ b/web/src/types/workbench.ts @@ -5,6 +5,8 @@ export type StageKey = 'V0.1' | 'V0.5' | 'V1.0'; export interface ProjectSession { sessionId: string; + /** 会话归属用户:每人与智能体私聊,互不可见 */ + userId: number; projectId: number; title: string; deviceType: string; @@ -73,6 +75,26 @@ export interface InteractiveCardData { resolvedBy?: string; } +export type AssistantMessageStatus = + | 'pending' + | 'streaming_status' + | 'auto_retrying' + | 'completed' + | 'failed' + | 'cancelled'; + +export interface ChatErrorInfo { + userMessage: string; + code: string; + class: 'transient' | 'resumable' | 'fatal' | 'cancelled'; + phaseId?: string; + upstream?: string; + requestId?: string; + sessionId?: string; + workspaceId?: string; + actions: Array<'retry' | 'continue' | 'restart'>; +} + export interface ChatMessage { id: string; sessionId: string; @@ -90,6 +112,11 @@ export interface ChatMessage { phaseEvents?: PhaseEvent[]; recommendedCases?: RecommendedCase[]; interactiveCard?: InteractiveCardData; + status?: AssistantMessageStatus; + livePhaseLabel?: string; + errorInfo?: ChatErrorInfo; + requestId?: string; + sourceUserMessageId?: string; } export type DeliverableCategory = 'workflow' | 'shared' | 'private';