diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts new file mode 100644 index 00000000..c036e22c --- /dev/null +++ b/frontend/src/entities/generation/api.test.ts @@ -0,0 +1,406 @@ +import { describe, expect, it, vi } from 'vitest' + +import { + createAuthenticatedGenerationTransport, + createGenerationApis, + GenerationApiError, +} from '@/entities' +import { EventStreamError } from '@/shared/api/stream' + +import type { MediaReference } from '../media' + +const reference = (url: string) => url as MediaReference + +function success(data: unknown): Response { + return new Response(JSON.stringify({ code: 200, message: 'success', data }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) +} + +function taskData(overrides: Record = {}) { + return { + id: 91, + user_id: 7, + project_id: 42, + task_type: 'character_image', + status: 'completed', + input_payload: { num_images: 4 }, + result: { + type: 'character_image', + image_urls: [ + 'https://cdn.test/candidate-1.png', + 'https://cdn.test/candidate-2.png', + 'https://cdn.test/candidate-3.png', + 'https://cdn.test/candidate-4.png', + ], + }, + error_message: null, + ...overrides, + } +} + +function actionFrames(count: number) { + return Array.from({ length: count }, (_, offset) => { + const index = count - offset - 1 + return { + index, + image_url: `https://cdn.test/frame-${index + 1}.png`, + duration_ms: index % 2 === 0 ? 100 : null, + } + }) +} + +describe('createGenerationApis', () => { + it('固定请求并映射四张角色母版候选', async () => { + const request = vi.fn(async (_url: string, _init?: RequestInit) => success(taskData())) + const stream = vi.fn(() => vi.fn()) + const apis = createGenerationApis({ + baseUrl: 'https://api.test/', + userId: '7', + transport: { request, stream }, + }) + + const generation = await apis.create({ + type: 'character_template', + projectId: '42', + referenceMedia: [reference('https://cdn.test/reference.png')], + prompt: 'pixel hero', + spriteWidth: 64, + spriteHeight: 96, + }) + + expect(request).toHaveBeenCalledWith( + 'https://api.test/generation/image', + expect.objectContaining({ + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + project_id: 42, + reference_image_url: 'https://cdn.test/reference.png', + prompt: 'pixel hero', + negative_prompt: '', + width: 64, + height: 96, + num_images: 4, + }), + }), + ) + expect(generation.result).toEqual({ + type: 'character_template', + images: [ + { url: 'https://cdn.test/candidate-1.png' }, + { url: 'https://cdn.test/candidate-2.png' }, + { url: 'https://cdn.test/candidate-3.png' }, + { url: 'https://cdn.test/candidate-4.png' }, + ], + }) + }) + + it('通过动作生成接口固定请求并映射一帧动作首帧', async () => { + const request = vi.fn(async (_url: string, _init?: RequestInit) => + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 1, action_type: 'idle' }, + result: { + type: 'character_action', + action_type: 'idle', + frames: [ + { index: 0, image_url: 'https://cdn.test/first-frame.png', duration_ms: null }, + ], + }, + }), + ), + ) + const apis = createGenerationApis({ + baseUrl: '', + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + const generation = await apis.create({ + type: 'first_frame', + projectId: '42', + characterId: '5', + outfitId: 'default', + actionType: 'idle', + prompt: 'stand naturally', + referenceMedia: [reference('https://cdn.test/template.png')], + }) + + expect(request.mock.calls[0]?.[0]).toBe('/generation/action') + expect(JSON.parse(String(request.mock.calls[0]?.[1]?.body))).toEqual({ + project_id: 42, + character_id: 5, + action_type: 'idle', + custom_prompt: 'stand naturally', + reference_video_url: null, + reference_image_urls: ['https://cdn.test/template.png'], + num_frames: 1, + }) + expect(generation.result).toEqual({ + type: 'first_frame', + image: { url: 'https://cdn.test/first-frame.png' }, + }) + }) + + it('以首帧请求完整动画并按后端 index 排序，当前合同固定为三十二帧', async () => { + const request = vi.fn(async (_url: string, _init?: RequestInit) => + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 32, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(32), + }, + }), + ), + ) + const apis = createGenerationApis({ + baseUrl: '/api', + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + const generation = await apis.create({ + type: 'complete_animation', + projectId: '42', + characterId: '5', + outfitId: 'default', + actionType: 'walk', + firstFrameUrl: 'https://cdn.test/frame-1.png', + prompt: 'move forward', + referenceMedia: [reference('https://cdn.test/extra.png')], + }) + + expect(request.mock.calls[0]?.[0]).toBe('/api/generation/action') + expect(JSON.parse(String(request.mock.calls[0]?.[1]?.body))).toEqual({ + project_id: 42, + character_id: 5, + action_type: 'walk', + custom_prompt: 'move forward', + reference_video_url: null, + reference_image_urls: ['https://cdn.test/frame-1.png', 'https://cdn.test/extra.png'], + num_frames: 32, + }) + expect(generation.result).toEqual({ + type: 'complete_animation', + frames: Array.from({ length: 32 }, (_, index) => ({ + url: `https://cdn.test/frame-${index + 1}.png`, + durationMs: index % 2 === 0 ? 100 : null, + })), + }) + }) + + it('拒绝未知任务状态而不是默认为 pending', async () => { + const request = vi.fn(async () => success(taskData({ status: 'queued' }))) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91')).rejects.toBeInstanceOf(GenerationApiError) + await expect(apis.get('42', '91')).rejects.toThrow('生成任务状态无效') + }) + + it('拒绝结果字段不完整的 completed DTO', async () => { + const request = vi.fn(async () => + success(taskData({ result: { type: 'character_image', image_urls: [null] } })), + ) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91')).rejects.toThrow('角色图片结果 image_urls 无效') + }) + + it('订阅 task_update，映射终态并把终态关闭信号交给流传输层', () => { + let subscribedUrl = '' + let streamOptions: + | { + eventName: string + onEvent(data: string): boolean + onError(error: Error): void + } + | undefined + const cancel = vi.fn() + const stream = vi.fn((url: string, options: NonNullable) => { + subscribedUrl = url + streamOptions = options + return cancel + }) + const apis = createGenerationApis({ + baseUrl: 'https://api.test', + userId: 7, + transport: { request: vi.fn(), stream }, + }) + const onEvent = vi.fn() + const onError = vi.fn() + + const unsubscribe = apis.subscribe('42', '91', onEvent, onError) + const isTerminal = streamOptions?.onEvent( + JSON.stringify({ + id: 91, + user_id: 7, + project_id: 42, + task_type: 'character_action', + status: 'completed', + input_payload: { num_frames: 32, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(32), + }, + error_message: null, + }), + ) + + expect(subscribedUrl).toBe('https://api.test/generation/tasks/91/stream?project_id=42') + expect(streamOptions?.eventName).toBe('task_update') + expect(isTerminal).toBe(true) + expect(onEvent).toHaveBeenCalledWith({ + taskId: '91', + type: 'complete_animation', + status: 'completed', + result: { + type: 'complete_animation', + frames: Array.from({ length: 32 }, (_, index) => ({ + url: `https://cdn.test/frame-${index + 1}.png`, + durationMs: index % 2 === 0 ? 100 : null, + })), + }, + error: null, + }) + + unsubscribe() + expect(cancel).toHaveBeenCalledOnce() + }) + + it('后端尚未提供 SSE 路由时退回任务查询，仍能交付终态', async () => { + const request = vi + .fn() + .mockResolvedValueOnce( + success(taskData({ status: 'running', result: null, error_message: null })), + ) + .mockResolvedValueOnce(success(taskData())) + const stream = vi.fn((_url, options) => { + queueMicrotask(() => + options.onError(new EventStreamError('SSE 请求失败（HTTP 404）', false, undefined, 404)), + ) + return vi.fn() + }) + const apis = createGenerationApis({ + userId: 7, + pollIntervalMs: 0, + transport: { request, stream }, + }) + const onEvent = vi.fn() + const onError = vi.fn() + + apis.subscribe('42', '91', onEvent, onError) + + await vi.waitFor(() => + expect(onEvent).toHaveBeenCalledWith( + expect.objectContaining({ + taskId: '91', + status: 'completed', + }), + ), + ) + expect(request).toHaveBeenCalledTimes(2) + expect(onError).not.toHaveBeenCalled() + }) + + it('拒绝 completed 任务返回错误动作类型', async () => { + const request = vi.fn(async () => + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 32, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'attack', + frames: actionFrames(32), + }, + }), + ), + ) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91')).rejects.toThrow('动作结果类型 attack 与请求的 walk 不一致') + }) + + it('拒绝不足三十二帧以及非失败状态携带错误', async () => { + const request = vi + .fn() + .mockResolvedValueOnce( + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 32, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(3), + }, + }), + ), + ) + .mockResolvedValueOnce(success(taskData({ error_message: 'provider failed' }))) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91')).rejects.toThrow('完整动画结果必须包含 32 帧') + await expect(apis.get('42', '91')).rejects.toThrow('completed 任务不应携带 error_message') + }) +}) + +describe('createAuthenticatedGenerationTransport', () => { + it('为普通请求携带 token，并在业务 401 后刷新和重放一次', async () => { + const requests: Request[] = [] + const fetchFn = vi + .fn() + .mockImplementationOnce(async (input, init) => { + requests.push(new Request(input, init)) + return new Response(JSON.stringify({ code: 401, message: 'expired', data: null }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) + }) + .mockImplementationOnce(async (input, init) => { + requests.push(new Request(input, init)) + return success(taskData({ status: 'pending', result: null })) + }) + const getAccessToken = vi + .fn<() => string>() + .mockReturnValueOnce('expired-token') + .mockReturnValueOnce('refreshed-token') + const recoverUnauthorized = vi.fn(async () => true) + const transport = createAuthenticatedGenerationTransport({ + fetchFn, + getAccessToken, + recoverUnauthorized, + }) + + const response = await transport.request('https://api.test/generation/image', { + method: 'POST', + body: '{}', + }) + + expect(response.ok).toBe(true) + expect(recoverUnauthorized).toHaveBeenCalledOnce() + expect(requests.map((request) => request.headers.get('authorization'))).toEqual([ + 'Bearer expired-token', + 'Bearer refreshed-token', + ]) + }) +}) diff --git a/frontend/src/entities/generation/api.ts b/frontend/src/entities/generation/api.ts new file mode 100644 index 00000000..223bb4aa --- /dev/null +++ b/frontend/src/entities/generation/api.ts @@ -0,0 +1,622 @@ +import { getApiAccessToken, recoverApiUnauthorized } from '@/shared/api' +import { + createEventStreamSubscriber, + EventStreamError, + type EventStreamSubscriber, +} from '@/shared/api/stream' + +import type { + CompleteAnimationGenerationInput, + GeneratedImage, + Generation, + GenerationApis, + GenerationEvent, + GenerationExpectation, + GenerationInput, + GenerationResult, + GenerationType, + TaskStatus, +} from '.' + +type RequestFunction = (url: string, init?: RequestInit) => Promise + +/** Generation 适配器需要的全部网络能力，由宿主统一注入。 */ +export interface GenerationTransport { + request: RequestFunction + stream: EventStreamSubscriber +} + +export interface AuthenticatedGenerationTransportOptions { + fetchFn?: typeof fetch + getAccessToken?: () => string | null | undefined + recoverUnauthorized?: () => Promise + reconnectDelayMs?: number +} + +export interface GenerationApiConfig { + /** API 前缀；空字符串表示同源。 */ + baseUrl?: string + /** 当前用户由认证宿主提供，适配器不猜测也不写死身份。 */ + userId: string | number + transport: GenerationTransport + /** SSE 路由尚未部署时，查询任务状态的间隔；测试可设为 0。 */ + pollIntervalMs?: number +} + +async function responseIsUnauthorized(response: Response): Promise { + if (response.status === 401) return true + if (!response.headers.get('content-type')?.includes('application/json')) return false + try { + const body = (await response.clone().json()) as unknown + return isRecord(body) && body.code === 401 + } catch { + return false + } +} + +/** + * Generation 需要读取原始响应并保持 SSE 长连接，不能直接套用会解包 JSON 的 ApiClient。 + * 这个传输器仍复用全局 token 与 401 恢复边界，避免生成模块另存一套登录态。 + */ +export function createAuthenticatedGenerationTransport( + options: AuthenticatedGenerationTransportOptions = {}, +): GenerationTransport { + const fetchFn = options.fetchFn ?? globalThis.fetch + const getAccessToken = options.getAccessToken ?? getApiAccessToken + const recoverUnauthorized = options.recoverUnauthorized ?? recoverApiUnauthorized + + return { + async request(url, init) { + let replayed = false + while (true) { + const headers = new Headers(init?.headers) + const accessToken = getAccessToken() + if (accessToken && !headers.has('authorization')) { + headers.set('authorization', `Bearer ${accessToken}`) + } + const response = await fetchFn(url, { + ...init, + headers, + credentials: init?.credentials ?? 'include', + }) + if (!replayed && (await responseIsUnauthorized(response))) { + replayed = true + if (await recoverUnauthorized()) continue + } + return response + } + }, + stream: createEventStreamSubscriber({ + fetchFn, + getAccessToken, + recoverUnauthorized, + reconnectDelayMs: options.reconnectDelayMs, + }), + } +} + +interface ResponseEnvelope { + code: unknown + message: unknown + data: unknown +} + +interface GenerationTaskDto { + id: number + userId: number + projectId: number + taskType: BackendGenerationType + status: TaskStatus + inputPayload: Record | null + result: Record | null + errorMessage: string | null +} + +type BackendGenerationType = 'character_image' | 'character_action' + +const TASK_STATUSES = new Set(['pending', 'running', 'completed', 'failed']) +export class GenerationApiError extends Error { + readonly code: number + + constructor(message: string, code = 0, options?: ErrorOptions) { + super(message, options) + this.name = 'GenerationApiError' + this.code = code + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function inputPositiveInteger(value: string | number, field: string): number { + const parsed = typeof value === 'number' ? value : Number(value) + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new GenerationApiError(`${field} 必须是正整数`) + } + return parsed +} + +function dtoPositiveInteger(value: unknown, field: string): number { + if (!Number.isSafeInteger(value) || (value as number) <= 0) { + throw new GenerationApiError(`生成任务 ${field} 无效`, 200) + } + return value as number +} + +function dtoNullableRecord(value: unknown, field: string): Record | null { + if (value === null) return null + if (!isRecord(value)) throw new GenerationApiError(`生成任务 ${field} 无效`, 200) + return value +} + +function dtoNullableString(value: unknown, field: string): string | null { + if (value === null) return null + if (typeof value !== 'string') throw new GenerationApiError(`生成任务 ${field} 无效`, 200) + return value +} + +function backendTaskType(value: unknown): BackendGenerationType { + if (value !== 'character_image' && value !== 'character_action') { + throw new GenerationApiError('生成任务 task_type 无效', 200) + } + return value +} + +function taskStatus(value: unknown): TaskStatus { + if (typeof value !== 'string' || !TASK_STATUSES.has(value as TaskStatus)) { + throw new GenerationApiError('生成任务状态无效', 200) + } + return value as TaskStatus +} + +function endpoint(baseUrl: string | undefined, path: string): string { + return `${(baseUrl ?? '').replace(/\/$/u, '')}${path}` +} + +async function readData(response: Response): Promise { + let raw: unknown + try { + raw = await response.json() + } catch (error) { + throw new GenerationApiError( + `生成接口返回了无法解析的响应（HTTP ${response.status}）`, + response.status, + { cause: error }, + ) + } + if (!isRecord(raw)) { + throw new GenerationApiError('生成接口响应不是对象', response.status) + } + + const envelope: ResponseEnvelope = { + code: raw.code, + message: raw.message, + data: raw.data, + } + if (typeof envelope.code !== 'number') { + throw new GenerationApiError('生成接口响应缺少有效的 code', response.status) + } + const message = + typeof envelope.message === 'string' ? envelope.message : `HTTP ${response.status}` + if (!response.ok || envelope.code !== 200) { + throw new GenerationApiError(message, envelope.code) + } + if (envelope.data === null || envelope.data === undefined) { + throw new GenerationApiError('生成接口成功响应缺少 data', envelope.code) + } + return envelope.data +} + +/** 完整查询 DTO 的每个字段都在网络边界校验，不把脏数据带入实体。 */ +function parseTaskDto(value: unknown): GenerationTaskDto { + if (!isRecord(value)) throw new GenerationApiError('生成任务响应不是对象', 200) + const inputPayload = dtoNullableRecord(value.input_payload, 'input_payload') + return { + id: dtoPositiveInteger(value.id, 'id'), + userId: dtoPositiveInteger(value.user_id, 'user_id'), + projectId: dtoPositiveInteger(value.project_id, 'project_id'), + taskType: backendTaskType(value.task_type), + status: taskStatus(value.status), + inputPayload, + result: dtoNullableRecord(value.result, 'result'), + errorMessage: dtoNullableString(value.error_message, 'error_message'), + } +} + +function expectedBackendType(type: GenerationType): BackendGenerationType { + return type === 'character_template' ? 'character_image' : 'character_action' +} + +function nonEmptyString(value: unknown, field: string): string { + if (typeof value !== 'string' || value.trim() === '') { + throw new GenerationApiError(`${field} 无效`, 200) + } + return value +} + +function mapImageResult(result: Record): GenerationResult { + if (result.type !== 'character_image') { + throw new GenerationApiError('角色图片结果 type 无效', 200) + } + if ( + !Array.isArray(result.image_urls) || + result.image_urls.length === 0 || + result.image_urls.some((url) => typeof url !== 'string' || url.trim() === '') + ) { + throw new GenerationApiError('角色图片结果 image_urls 无效', 200) + } + const images = result.image_urls.map((url): GeneratedImage => ({ url: url as string })) + + if (images.length !== 4) { + throw new GenerationApiError('角色母版结果必须包含 4 个候选', 200) + } + return { type: 'character_template', images } +} + +function mapActionResult( + result: Record, + expectation: Extract, +): GenerationResult { + if (result.type !== 'character_action') { + throw new GenerationApiError('完整动画结果 type 无效', 200) + } + if (typeof result.action_type !== 'string' || result.action_type.trim() === '') { + throw new GenerationApiError('完整动画结果 action_type 无效', 200) + } + if (result.action_type !== expectation.actionType) { + throw new GenerationApiError( + `动作结果类型 ${result.action_type} 与请求的 ${expectation.actionType} 不一致`, + 200, + ) + } + if (!Array.isArray(result.frames) || result.frames.length === 0) { + throw new GenerationApiError('完整动画结果 frames 无效', 200) + } + + const indexes = new Set() + const frames = result.frames.map((frame) => { + if (!isRecord(frame)) throw new GenerationApiError('动作帧不是对象', 200) + if (!Number.isSafeInteger(frame.index) || (frame.index as number) < 0) { + throw new GenerationApiError('动作帧 index 无效', 200) + } + const index = frame.index as number + if (indexes.has(index)) throw new GenerationApiError('动作帧 index 重复', 200) + indexes.add(index) + if ( + frame.duration_ms !== null && + (!Number.isFinite(frame.duration_ms) || (frame.duration_ms as number) < 0) + ) { + throw new GenerationApiError('动作帧 duration_ms 无效', 200) + } + return { + index, + url: nonEmptyString(frame.image_url, '动作帧 image_url'), + durationMs: frame.duration_ms as number | null, + } + }) + + const orderedFrames = frames.sort((left, right) => left.index - right.index) + const expectedFrameCount = expectation.type === 'first_frame' ? 1 : 32 + if (orderedFrames.length !== expectedFrameCount) { + throw new GenerationApiError( + `${expectation.type === 'first_frame' ? '动作首帧' : '完整动画'}结果必须包含 ${expectedFrameCount} 帧`, + 200, + ) + } + for (let index = 0; index < expectedFrameCount; index += 1) { + if (!indexes.has(index)) { + throw new GenerationApiError('动作帧 index 必须从 0 开始连续排列', 200) + } + } + if (expectation.type === 'first_frame') { + return { type: 'first_frame', image: { url: orderedFrames[0]!.url } } + } + return { + type: 'complete_animation', + frames: orderedFrames.map(({ url, durationMs }) => ({ url, durationMs })), + } +} + +function mapResult( + result: Record | null, + status: TaskStatus, + expectation: GenerationExpectation, +): GenerationResult | null { + if (status !== 'completed') { + if (result !== null) { + throw new GenerationApiError('非完成任务不应携带 result', 200) + } + return null + } + if (result === null) throw new GenerationApiError('完成任务缺少 result', 200) + return expectation.type === 'character_template' + ? mapImageResult(result) + : mapActionResult(result, expectation) +} + +function validateStatusError(status: TaskStatus, error: string | null): void { + if (status === 'failed') { + if (error === null || error.trim() === '') { + throw new GenerationApiError('失败任务缺少 error_message', 200) + } + return + } + if (error !== null) { + throw new GenerationApiError(`${status} 任务不应携带 error_message`, 200) + } +} + +function validateInputPayload( + inputPayload: Record | null, + expectation: GenerationExpectation, +): void { + if (inputPayload === null) { + throw new GenerationApiError('生成任务缺少 input_payload', 200) + } + if (expectation.type === 'character_template') { + if (inputPayload.num_images !== 4) { + throw new GenerationApiError('角色母版任务 input_payload.num_images 必须为 4', 200) + } + return + } + const expectedFrameCount = expectation.type === 'first_frame' ? 1 : 32 + if (inputPayload.num_frames !== expectedFrameCount) { + throw new GenerationApiError( + `动作任务 input_payload.num_frames 必须为 ${expectedFrameCount}`, + 200, + ) + } + if (inputPayload.action_type !== expectation.actionType) { + throw new GenerationApiError('动作任务 input_payload.action_type 与请求不一致', 200) + } +} + +function validateTaskIdentity( + dto: GenerationTaskDto, + expectedProjectId: number, + expectedUserId: number, + expectation: GenerationExpectation, + expectedTaskId?: number, +): void { + if (dto.projectId !== expectedProjectId) { + throw new GenerationApiError(`生成任务未归属请求中的项目 ${expectedProjectId}`, 200) + } + if (dto.userId !== expectedUserId) { + throw new GenerationApiError('生成任务未归属当前用户', 200) + } + if (expectedTaskId !== undefined && dto.id !== expectedTaskId) { + throw new GenerationApiError(`生成任务 ID 与请求的 ${expectedTaskId} 不一致`, 200) + } + if (dto.taskType !== expectedBackendType(expectation.type)) { + throw new GenerationApiError(`生成任务类型与 ${expectation.type} 不匹配`, 200) + } + validateStatusError(dto.status, dto.errorMessage) + validateInputPayload(dto.inputPayload, expectation) +} + +function deriveExpectation(dto: GenerationTaskDto): GenerationExpectation { + if (dto.inputPayload === null) { + throw new GenerationApiError('生成任务缺少 input_payload', 200) + } + if (dto.taskType === 'character_image') return { type: 'character_template' } + + const actionType = nonEmptyString(dto.inputPayload.action_type, '动作任务 action_type') + if (dto.inputPayload.num_frames === 1) return { type: 'first_frame', actionType } + if (dto.inputPayload.num_frames === 32) return { type: 'complete_animation', actionType } + throw new GenerationApiError('动作任务 input_payload.num_frames 必须为 1 或 32', 200) +} + +function sameExpectation(left: GenerationExpectation, right: GenerationExpectation): boolean { + if (left.type !== right.type) return false + if (left.type === 'character_template' || right.type === 'character_template') return true + return left.actionType === right.actionType +} + +function mapTask( + value: unknown, + expectedProjectId: number, + expectedUserId: number, + expectedExpectation?: GenerationExpectation, + expectedTaskId?: number, +): Generation { + const dto = parseTaskDto(value) + const expectation = deriveExpectation(dto) + if (expectedExpectation && !sameExpectation(expectation, expectedExpectation)) { + throw new GenerationApiError(`生成任务类型与 ${expectedExpectation.type} 不匹配`, 200) + } + validateTaskIdentity(dto, expectedProjectId, expectedUserId, expectation, expectedTaskId) + return { + id: String(dto.id), + projectId: String(dto.projectId), + type: expectation.type, + status: dto.status, + result: mapResult(dto.result, dto.status, expectation), + error: dto.errorMessage, + } +} + +function references(input: CompleteAnimationGenerationInput): string[] { + return [input.firstFrameUrl, ...input.referenceMedia.map(String)].filter( + (url, index, all) => url.trim() !== '' && all.indexOf(url) === index, + ) +} + +function parseEventData(data: string): unknown { + try { + return JSON.parse(data) as unknown + } catch (error) { + throw new GenerationApiError('task_update 不是有效 JSON', 200, { cause: error }) + } +} + +function mapEvent( + value: unknown, + expectedProjectId: number, + expectedUserId: number, + expectedTaskId: number, +): GenerationEvent { + const generation = mapTask(value, expectedProjectId, expectedUserId, undefined, expectedTaskId) + return { + taskId: generation.id, + type: generation.type, + status: generation.status, + result: generation.result, + error: generation.error, + } +} + +/** + * 创建 Generation 实体适配器。 + * + * `userId` 与 HTTP/SSE transport 都由宿主注入，因此模块既不持有登录态，也不直接 + * 依赖 fetch/EventSource。三个前端阶段在这里收口为后端的两类 GenerationTask。 + */ +export function createGenerationApis(config: GenerationApiConfig): GenerationApis { + const userId = inputPositiveInteger(config.userId, 'userId') + const { request, stream } = config.transport + const pollIntervalMs = config.pollIntervalMs ?? 1_000 + + async function post( + path: '/generation/image' | '/generation/action', + projectId: number, + expectation: Extract, + body: Record, + ): Promise> { + const response = await request(endpoint(config.baseUrl, path), { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }) + return mapTask(await readData(response), projectId, userId, expectation) as Generation + } + + async function getTask(projectId: string, id: string): Promise { + const numericProjectId = inputPositiveInteger(projectId, 'projectId') + const numericTaskId = inputPositiveInteger(id, 'taskId') + const response = await request( + endpoint(config.baseUrl, `/generation/tasks/${numericTaskId}?project_id=${numericProjectId}`), + { method: 'GET' }, + ) + return mapTask(await readData(response), numericProjectId, userId, undefined, numericTaskId) + } + + return { + async create(input: T): Promise> { + const projectId = inputPositiveInteger(input.projectId, 'projectId') + if (input.type !== 'character_template') { + const referenceImageUrls = + input.type === 'complete_animation' + ? references(input) + : input.referenceMedia.map(String).filter((url) => url.trim() !== '') + return post( + '/generation/action', + projectId, + { type: input.type, actionType: input.actionType }, + { + project_id: projectId, + character_id: inputPositiveInteger(input.characterId, 'characterId'), + action_type: input.actionType, + custom_prompt: input.prompt, + reference_video_url: null, + reference_image_urls: referenceImageUrls, + // 首帧是一帧动作任务；完整动画按产品合同固定生成 32 帧。 + num_frames: input.type === 'first_frame' ? 1 : 32, + }, + ) + } + + return post( + '/generation/image', + projectId, + { type: input.type }, + { + project_id: projectId, + reference_image_url: input.referenceMedia[0] ? String(input.referenceMedia[0]) : null, + prompt: input.prompt ?? '', + negative_prompt: '', + width: inputPositiveInteger(input.spriteWidth, 'spriteWidth'), + height: inputPositiveInteger(input.spriteHeight, 'spriteHeight'), + // 只有角色母版走图片接口，并且固定生成四个候选。 + num_images: 4, + }, + ) + }, + + get: getTask, + + subscribe( + projectId: string, + id: string, + onEvent: (event: GenerationEvent) => void, + onError = () => undefined, + ): () => void { + const numericProjectId = inputPositiveInteger(projectId, 'projectId') + const numericTaskId = inputPositiveInteger(id, 'taskId') + let active = true + let polling = false + let timer: ReturnType | null = null + let wakePoll: (() => void) | null = null + const stopStream = stream( + endpoint( + config.baseUrl, + `/generation/tasks/${numericTaskId}/stream?project_id=${numericProjectId}`, + ), + { + eventName: 'task_update', + onEvent(data) { + const event = mapEvent(parseEventData(data), numericProjectId, userId, numericTaskId) + onEvent(event) + return event.status === 'completed' || event.status === 'failed' + }, + onError(error) { + if (error instanceof EventStreamError && error.status === 404) { + polling = true + void poll() + return + } + onError(error) + }, + }, + ) + + async function poll(): Promise { + while (active && polling) { + try { + const generation = await getTask(String(numericProjectId), String(numericTaskId)) + const event: GenerationEvent = { + taskId: generation.id, + type: generation.type, + status: generation.status, + result: generation.result, + error: generation.error, + } + onEvent(event) + if (event.status === 'completed' || event.status === 'failed') return + } catch (error) { + if (active) onError(error instanceof Error ? error : new Error(String(error))) + return + } + await new Promise((resolve) => { + wakePoll = resolve + timer = setTimeout(() => { + wakePoll = null + resolve() + }, pollIntervalMs) + }) + timer = null + } + } + + return () => { + if (!active) return + active = false + polling = false + if (timer !== null) clearTimeout(timer) + wakePoll?.() + wakePoll = null + stopStream() + } + }, + } +} diff --git a/frontend/src/entities/generation/index.ts b/frontend/src/entities/generation/index.ts index 33f21313..8be61066 100644 --- a/frontend/src/entities/generation/index.ts +++ b/frontend/src/entities/generation/index.ts @@ -19,25 +19,38 @@ export type TaskStatus = 'pending' | 'running' | 'completed' | 'failed' /** * 生成对应的三个前端可见异步步骤。 * 它是前端工作流粒度，不等于后端 task_type——后端只有 character_image 与 - * character_action 两种，character_template 和 first_frame 都落在 character_image 上。 + * character_action 两种：character_template 落在 character_image，动作首帧和完整动画 + * 都落在 character_action，只是请求帧数分别为 1 和 32。 * 完整动画内部可含视频生成、截帧和多次图像处理，但对前端仍是一次 Generation。 */ export type GenerationType = 'character_template' | 'first_frame' | 'complete_animation' +/** + * 恢复任务时由 WorkflowRun 提供的已知上下文。动作阶段必须带上动作语义， + * 这样适配器才能拒绝“请求 walk、后端却返回 attack”这类串任务结果。 + */ +export type GenerationExpectation = + | { type: 'character_template' } + | { type: 'first_frame'; actionType: ActionType } + | { type: 'complete_animation'; actionType: ActionType } + interface GenerationInputBase { projectId: string /** 可选参考媒体；没有参考图时传空数组。 */ referenceMedia: readonly MediaReference[] } -/** 角色母版候选生成。 */ +/** 角色母版候选生成；当前合同固定请求 4 个候选，不向调用方暴露可变数量。 */ export interface CharacterTemplateGenerationInput extends GenerationInputBase { type: 'character_template' /** 已由手动输入或 Quick Start 整理好的角色提示词。 */ prompt: string + /** 必须与 Project 的精灵尺寸一致，由上层用例明确传入。 */ + spriteWidth: number + spriteHeight: number } -/** 指定角色造型下的动作首帧生成；不能只绑定 Character。 */ +/** 指定角色造型下的动作首帧生成；当前合同固定 1 张，且不能只绑定 Character。 */ export interface FirstFrameGenerationInput extends GenerationInputBase { type: 'first_frame' characterId: string @@ -47,7 +60,10 @@ export interface FirstFrameGenerationInput extends GenerationInputBase { prompt: string | null } -/** 以已确认首帧为起点生成完整动画。 */ +/** + * 以已确认首帧为起点生成完整动画。 + * 产品合同固定生成 32 帧；适配器不再沿用后端旧默认值 16。 + */ export interface CompleteAnimationGenerationInput extends GenerationInputBase { type: 'complete_animation' characterId: string @@ -68,6 +84,11 @@ export interface GeneratedImage { url: string } +/** 后端动作结果中的一帧；null 时由 Action.fps 提供等时长回退。 */ +export interface GeneratedFrame extends GeneratedImage { + durationMs: number | null +} + /** 结果按 type 分别定义，不共用一个 urls 数组。 */ export interface CharacterTemplateGenerationResult { type: 'character_template' @@ -79,10 +100,10 @@ export interface FirstFrameGenerationResult { image: GeneratedImage } -/** 帧顺序由数组位置表达。 */ +/** 帧顺序由数组位置表达，同时保留后端逐帧时长。 */ export interface CompleteAnimationGenerationResult { type: 'complete_animation' - frames: readonly GeneratedImage[] + frames: readonly GeneratedFrame[] } export type GenerationResult = @@ -101,7 +122,8 @@ export type GenerationResultFor = * 一次生成任务的完整快照，创建、查询和断线恢复都用它。 * 它是服务端的资源，不是一次「调用能力」——前端创建它，然后订阅或轮询它的状态。 * - * TType 在调用边界已知时保留精确类型；按 ID 恢复时用默认值，等运行时解析后再收窄。 + * TType 在调用边界已知时保留精确类型；查询和订阅根据持久化 input_payload + * 还原首帧或完整动画阶段，因为后端 task_type 比前端阶段更粗。 * 完成不代表工作流节点已通过，节点状态由 WorkflowNode 自己判定。 */ export interface Generation { @@ -117,15 +139,14 @@ export interface Generation { } /** - * 一条状态变更事件。 - * 不含 projectId：后端事件 payload 只有 task_id、task_type、status， - * 以及完成时的 result 和失败时的 error_message。 + * 一条状态变更事件。后端 SSE 推送完整任务对象并使用 id；适配器校验后将 id + * 转成 taskId，避免传输字段名泄漏到 Controller。 */ export interface GenerationEvent extends Omit< Generation, 'id' | 'projectId' > { - /** 对应 Generation.id，字段名沿用后端事件里的 task_id。 */ + /** 对应后端事件的 id，也对应 Generation.id。 */ taskId: Generation['id'] } @@ -138,10 +159,25 @@ export interface GenerationApis { * projectId 不能从 id 推导，后端查询接口要求两者同时传入。 */ get(projectId: Generation['projectId'], id: Generation['id']): Promise - /** 订阅状态变化，返回取消订阅函数。 */ + /** + * 订阅 task_update，终态由传输层自动关闭；返回的函数供页面离开时主动取消。 + * 传输错误与非法 DTO 通过 onError 上报，不伪造成业务 failed 状态。 + */ subscribe( projectId: Generation['projectId'], id: Generation['id'], onEvent: (event: GenerationEvent) => void, + onError?: (error: Error) => void, ): () => void } + +export { + createAuthenticatedGenerationTransport, + createGenerationApis, + GenerationApiError, +} from './api' +export type { + AuthenticatedGenerationTransportOptions, + GenerationApiConfig, + GenerationTransport, +} from './api' diff --git a/frontend/src/entities/index.ts b/frontend/src/entities/index.ts index 4125ea91..1bf57ca5 100644 --- a/frontend/src/entities/index.ts +++ b/frontend/src/entities/index.ts @@ -31,7 +31,12 @@ export { characterApis } from './character' /* 动作模板 —— 能跨角色复用的配方 */ export type { ActionTemplate, ActionTemplateApis } from './action-template' -/* 生成 —— 业务数据，不是「调用生成能力」 */ +/* 生成 —— 业务数据，不是「调用生成能力」；后端的 task 就是它，不另立实体 */ +export { + createAuthenticatedGenerationTransport, + createGenerationApis, + GenerationApiError, +} from './generation' export type { CharacterTemplateGenerationInput, CharacterTemplateGenerationResult, @@ -39,14 +44,19 @@ export type { CompleteAnimationGenerationResult, FirstFrameGenerationInput, FirstFrameGenerationResult, + GeneratedFrame, GeneratedImage, Generation, GenerationApis, GenerationEvent, + GenerationExpectation, GenerationInput, GenerationResult, GenerationResultFor, GenerationType, + GenerationApiConfig, + AuthenticatedGenerationTransportOptions, + GenerationTransport, TaskStatus, } from './generation' diff --git a/frontend/src/shared/api/index.ts b/frontend/src/shared/api/index.ts index 9ef07dda..ccac1f0a 100644 --- a/frontend/src/shared/api/index.ts +++ b/frontend/src/shared/api/index.ts @@ -67,6 +67,17 @@ function getApiUnauthorizedRecovery(): ApiUnauthorizedRecovery | undefined { return unauthorizedRecoveryProviders.at(-1) } +/** 供需要原始 Response 或流式响应的适配器复用同一套会话恢复能力。 */ +export async function recoverApiUnauthorized(): Promise { + const recovery = getApiUnauthorizedRecovery() + if (!recovery) return false + try { + return await recovery() + } catch { + return false + } +} + export type ApiErrorKind = 'business' | 'http' | 'invalid-response' | 'network' /** 后端业务错误与传输错误统一进入这一种前端错误。 */ @@ -227,21 +238,14 @@ export function createApiClient({ const response = await send(path, options) const envelope = await readEnvelope(response) - const recovery = getApiUnauthorizedRecovery() if ( !replayed && recoverUnauthorized && response.status === 200 && envelope.code === 401 && - recovery && canReplay(options) ) { - let recovered = false - try { - recovered = await recovery() - } catch { - // 恢复失败仍应向调用方交付原始 401，而不是泄漏 refresh 的错误。 - } + const recovered = await recoverApiUnauthorized() if (recovered) return receiveEnvelope(path, options, true) } diff --git a/frontend/src/shared/api/stream.test.ts b/frontend/src/shared/api/stream.test.ts new file mode 100644 index 00000000..e4ab9143 --- /dev/null +++ b/frontend/src/shared/api/stream.test.ts @@ -0,0 +1,167 @@ +import { describe, expect, it, vi } from 'vitest' + +import { createEventStreamSubscriber } from './stream' + +function eventStreamResponse(data: string, event = 'task_update'): Response { + return new Response(`event: ${event}\ndata: ${data}\n\n`, { + headers: { 'content-type': 'text/event-stream' }, + }) +} + +describe('createEventStreamSubscriber', () => { + it('使用 Bearer Token 建立 SSE 连接并在终态后停止', async () => { + let request: Request | undefined + const fetchFn = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + request = new Request(input, init) + return eventStreamResponse('{"status":"completed"}') + }) + const subscriber = createEventStreamSubscriber({ + fetchFn, + getAccessToken: () => 'access-token', + }) + + await new Promise((resolve, reject) => { + subscriber('https://api.test/generation/tasks/91/stream?project_id=42', { + eventName: 'task_update', + onEvent(data) { + expect(data).toBe('{"status":"completed"}') + resolve() + return true + }, + onError: reject, + }) + }) + + expect(request?.headers.get('accept')).toBe('text/event-stream') + expect(request?.headers.get('authorization')).toBe('Bearer access-token') + expect(fetchFn).toHaveBeenCalledOnce() + }) + + it('HTTP 401 时刷新会话并用新 token 重连一次', async () => { + const requests: Request[] = [] + const fetchFn = vi + .fn() + .mockImplementationOnce(async (input, init) => { + requests.push(new Request(input, init)) + return new Response(null, { status: 401 }) + }) + .mockImplementationOnce(async (input, init) => { + requests.push(new Request(input, init)) + return eventStreamResponse('{"status":"completed"}') + }) + const getAccessToken = vi + .fn<() => string>() + .mockReturnValueOnce('expired-token') + .mockReturnValueOnce('refreshed-token') + const recoverUnauthorized = vi.fn(async () => true) + const subscriber = createEventStreamSubscriber({ + fetchFn, + getAccessToken, + recoverUnauthorized, + reconnectDelayMs: 0, + }) + + await new Promise((resolve, reject) => { + subscriber('https://api.test/generation/tasks/91/stream', { + eventName: 'task_update', + onEvent() { + resolve() + return true + }, + onError: reject, + }) + }) + + expect(recoverUnauthorized).toHaveBeenCalledOnce() + expect(requests.map((request) => request.headers.get('authorization'))).toEqual([ + 'Bearer expired-token', + 'Bearer refreshed-token', + ]) + }) + + it('网络中断后重连，并为新连接重新读取 access token', async () => { + const getAccessToken = vi + .fn<() => string>() + .mockReturnValueOnce('first-token') + .mockReturnValueOnce('second-token') + const requests: Request[] = [] + const fetchFn = vi + .fn() + .mockRejectedValueOnce(new TypeError('connection reset')) + .mockImplementationOnce(async () => eventStreamResponse('{"status":"failed"}')) + const onError = vi.fn() + const subscriber = createEventStreamSubscriber({ + fetchFn: async (input, init) => { + requests.push(new Request(input, init)) + return fetchFn(input, init) + }, + getAccessToken, + reconnectDelayMs: 0, + }) + + await new Promise((resolve) => { + subscriber('https://api.test/generation/tasks/91/stream', { + eventName: 'task_update', + onEvent() { + resolve() + return true + }, + onError, + }) + }) + + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: 'SSE 连接中断，正在自动重连' }), + ) + expect(requests.map((request) => request.headers.get('authorization'))).toEqual([ + 'Bearer first-token', + 'Bearer second-token', + ]) + }) + + it('取消订阅会中止正在进行的请求', async () => { + let signal: AbortSignal | undefined + const subscriber = createEventStreamSubscriber({ + async fetchFn(_input, init) { + signal = init?.signal as AbortSignal + return new Promise(() => undefined) + }, + getAccessToken: () => undefined, + }) + + const unsubscribe = subscriber('https://api.test/generation/tasks/91/stream', { + eventName: 'task_update', + onEvent: () => false, + onError: vi.fn(), + }) + await vi.waitFor(() => expect(signal).toBeDefined()) + unsubscribe() + + expect(signal?.aborted).toBe(true) + }) + + it('业务事件解析失败时报告错误且不重连', async () => { + const fetchFn = vi.fn(async () => eventStreamResponse('{}')) + const onError = vi.fn() + const subscriber = createEventStreamSubscriber({ + fetchFn, + getAccessToken: () => undefined, + reconnectDelayMs: 0, + }) + + subscriber('https://api.test/generation/tasks/91/stream', { + eventName: 'task_update', + onEvent() { + throw new Error('invalid task DTO') + }, + onError, + }) + + await vi.waitFor(() => + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: 'invalid task DTO' }), + ), + ) + expect(fetchFn).toHaveBeenCalledOnce() + }) +}) diff --git a/frontend/src/shared/api/stream.ts b/frontend/src/shared/api/stream.ts new file mode 100644 index 00000000..33a78d8d --- /dev/null +++ b/frontend/src/shared/api/stream.ts @@ -0,0 +1,195 @@ +/** 业务无关的、可鉴权的 SSE 订阅边界。 */ + +export interface EventStreamOptions { + /** 只监听业务指定的命名事件，例如 task_update。 */ + eventName: string + /** 返回 true 表示 payload 是终态，传输层随后关闭连接。 */ + onEvent(data: string): boolean + /** 包含连接中断、非法响应和业务解析器抛出的错误。 */ + onError(error: Error): void +} + +export type EventStreamSubscriber = (url: string, options: EventStreamOptions) => () => void + +export interface EventStreamSubscriberConfig { + fetchFn?: typeof fetch + /** 每次连接前重新读取，支持刷新后的 token。 */ + getAccessToken: () => string | null | undefined + /** HTTP 401 时由认证会话尝试刷新；成功后只重放本次连接。 */ + recoverUnauthorized?: () => Promise + reconnectDelayMs?: number +} + +export class EventStreamError extends Error { + readonly retryable: boolean + readonly status: number | null + + constructor( + message: string, + retryable = false, + options?: ErrorOptions, + status: number | null = null, + ) { + super(message, options) + this.name = 'EventStreamError' + this.retryable = retryable + this.status = status + } +} + +interface SseRecord { + event: string + data: string +} + +function asError(value: unknown): Error { + return value instanceof Error ? value : new EventStreamError('SSE 事件处理失败') +} + +function connectionError(cause?: unknown): EventStreamError { + return new EventStreamError('SSE 连接中断，正在自动重连', true, { cause }) +} + +function parseRecord(block: string): SseRecord | null { + let event = 'message' + const data: string[] = [] + for (const line of block.split(/\r?\n/u)) { + if (line.startsWith(':')) continue + const separator = line.indexOf(':') + const field = separator < 0 ? line : line.slice(0, separator) + const rawValue = separator < 0 ? '' : line.slice(separator + 1) + const value = rawValue.startsWith(' ') ? rawValue.slice(1) : rawValue + if (field === 'event') event = value + if (field === 'data') data.push(value) + } + return data.length === 0 ? null : { event, data: data.join('\n') } +} + +async function readEventStream(response: Response, options: EventStreamOptions): Promise { + if (!response.body) throw new EventStreamError('SSE 响应缺少消息流') + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + + const deliver = async (block: string): Promise => { + const record = parseRecord(block) + if (record?.event !== options.eventName) return false + if (!options.onEvent(record.data)) return false + await reader.cancel() + return true + } + + try { + while (true) { + let chunk: ReadableStreamReadResult + try { + chunk = await reader.read() + } catch (cause) { + throw connectionError(cause) + } + const { done, value } = chunk + buffer += decoder.decode(value, { stream: !done }) + let boundary = /\r?\n\r?\n/u.exec(buffer) + while (boundary) { + const block = buffer.slice(0, boundary.index) + buffer = buffer.slice(boundary.index + boundary[0].length) + if (await deliver(block)) return true + boundary = /\r?\n\r?\n/u.exec(buffer) + } + if (!done) continue + return buffer.length > 0 ? deliver(buffer) : false + } + } catch (cause) { + try { + await reader.cancel(cause) + } catch { + // 取消失败不能覆盖真正的协议或业务错误。 + } + throw cause + } finally { + reader.releaseLock() + } +} + +function waitForReconnect(delayMs: number, signal: AbortSignal): Promise { + if (delayMs <= 0 || signal.aborted) return Promise.resolve() + return new Promise((resolve) => { + const finish = () => { + clearTimeout(timer) + signal.removeEventListener('abort', finish) + resolve() + } + const timer = setTimeout(finish, delayMs) + signal.addEventListener('abort', finish, { once: true }) + }) +} + +/** + * 使用 fetch 流建立 SSE。浏览器原生 EventSource 不能设置 Authorization， + * 因此不能用于当前受保护的任务订阅接口。 + */ +export function createEventStreamSubscriber( + config: EventStreamSubscriberConfig, +): EventStreamSubscriber { + const fetchFn = config.fetchFn ?? globalThis.fetch + const reconnectDelayMs = config.reconnectDelayMs ?? 1_000 + + return (url, options) => { + const controller = new AbortController() + let attemptedUnauthorizedRecovery = false + + const run = async () => { + while (!controller.signal.aborted) { + try { + const headers = new Headers({ Accept: 'text/event-stream' }) + const accessToken = config.getAccessToken() + if (accessToken) headers.set('Authorization', `Bearer ${accessToken}`) + let response: Response + try { + response = await fetchFn(url, { + method: 'GET', + headers, + credentials: 'include', + signal: controller.signal, + }) + } catch (cause) { + throw connectionError(cause) + } + + if ( + response.status === 401 && + !attemptedUnauthorizedRecovery && + config.recoverUnauthorized + ) { + attemptedUnauthorizedRecovery = true + if (await config.recoverUnauthorized()) continue + } + if (!response.ok) { + throw new EventStreamError( + `SSE 请求失败（HTTP ${response.status}）`, + false, + undefined, + response.status, + ) + } + if (!response.headers.get('content-type')?.includes('text/event-stream')) { + throw new EventStreamError('SSE 响应类型无效') + } + attemptedUnauthorizedRecovery = false + const terminal = await readEventStream(response, options) + if (terminal || controller.signal.aborted) return + throw connectionError() + } catch (cause) { + if (controller.signal.aborted) return + const error = asError(cause) + options.onError(error) + if (!(error instanceof EventStreamError) || !error.retryable) return + await waitForReconnect(reconnectDelayMs, controller.signal) + } + } + } + + void run() + return () => controller.abort() + } +}