From 3fcdceb5a41ee23d26194596f78a05b26e233e18 Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:57:39 +0800 Subject: [PATCH 1/5] feat(generation): add compatible authenticated SSE adapter --- frontend/src/entities/generation/api.test.ts | 380 +++++++++++++ frontend/src/entities/generation/api.ts | 557 +++++++++++++++++++ frontend/src/entities/generation/index.ts | 20 + frontend/src/entities/index.ts | 3 + frontend/src/shared/api/stream.test.ts | 167 ++++++ frontend/src/shared/api/stream.ts | 195 +++++++ 6 files changed, 1322 insertions(+) create mode 100644 frontend/src/entities/generation/api.test.ts create mode 100644 frontend/src/entities/generation/api.ts create mode 100644 frontend/src/shared/api/stream.test.ts create mode 100644 frontend/src/shared/api/stream.ts diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts new file mode 100644 index 00000000..767010f0 --- /dev/null +++ b/frontend/src/entities/generation/api.test.ts @@ -0,0 +1,380 @@ +import { describe, expect, it, vi } from 'vitest' + +import { createGenerationApis, GenerationApiError } from '@/entities' + +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', { type: 'character_template' })).rejects.toBeInstanceOf( + GenerationApiError, + ) + await expect(apis.get('42', '91', { type: 'character_template' })).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', { type: 'character_template' })).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', + { type: 'complete_animation', actionType: 'walk' }, + 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('从查询结果推断前端阶段，并允许现有三参数订阅继续使用', async () => { + let streamOptions: + | { + eventName: string + onEvent(data: string): boolean + onError(error: Error): void + } + | undefined + const task = taskData({ + task_type: 'character_action', + input_payload: { num_frames: 32, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(32), + }, + }) + const stream = vi.fn((_url: string, options: NonNullable) => { + streamOptions = options + return vi.fn() + }) + const apis = createGenerationApis({ + userId: 7, + transport: { request: vi.fn(async () => success(task)), stream }, + }) + + const generation = await apis.get('42', '91') + const onEvent = vi.fn() + apis.subscribe('42', '91', onEvent) + streamOptions?.onEvent(JSON.stringify(task)) + + expect(generation.type).toBe('complete_animation') + expect(onEvent).toHaveBeenCalledWith( + expect.objectContaining({ taskId: '91', type: 'complete_animation', status: 'completed' }), + ) + }) + + 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', { type: 'complete_animation', actionType: 'walk' }), + ).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', { type: 'complete_animation', actionType: 'walk' }), + ).rejects.toThrow('完整动画结果必须包含 32 帧') + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow( + 'completed 任务不应携带 error_message', + ) + }) +}) diff --git a/frontend/src/entities/generation/api.ts b/frontend/src/entities/generation/api.ts new file mode 100644 index 00000000..2f31b542 --- /dev/null +++ b/frontend/src/entities/generation/api.ts @@ -0,0 +1,557 @@ +import 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 GenerationApiConfig { + /** API 前缀；空字符串表示同源。 */ + baseUrl?: string + /** 当前用户由认证宿主提供，适配器不猜测也不写死身份。 */ + userId: string | number + transport: GenerationTransport +} + +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']) +const ACTION_TYPES = new Set(['walk', 'idle', 'attack', 'jump', 'custom']) + +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' || !ACTION_TYPES.has(result.action_type)) { + 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 inferExpectation(dto: GenerationTaskDto): GenerationExpectation { + if (dto.taskType === 'character_image') return { type: 'character_template' } + if (dto.inputPayload === null) { + throw new GenerationApiError('动作任务缺少 input_payload', 200) + } + const actionType = dto.inputPayload.action_type + if (typeof actionType !== 'string' || !ACTION_TYPES.has(actionType)) { + throw new GenerationApiError('动作任务 input_payload.action_type 无效', 200) + } + 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 无法映射到前端阶段', 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 mapTask( + value: unknown, + expectedProjectId: number, + expectedUserId: number, + expectation?: GenerationExpectation, + expectedTaskId?: number, +): Generation { + const dto = parseTaskDto(value) + const resolvedExpectation = expectation ?? inferExpectation(dto) + validateTaskIdentity(dto, expectedProjectId, expectedUserId, resolvedExpectation, expectedTaskId) + return { + id: String(dto.id), + projectId: String(dto.projectId), + type: resolvedExpectation.type, + status: dto.status, + result: mapResult(dto.result, dto.status, resolvedExpectation), + 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, + expectation: Extract, +): GenerationEvent { + if (!isRecord(value)) throw new GenerationApiError('task_update 不是对象', 200) + // 后端 PR #34 推送完整 GenerationTask，标识字段是数字 id；前端统一转为字符串 taskId。 + const taskId = dtoPositiveInteger(value.id, 'id') + if (taskId !== expectedTaskId) { + throw new GenerationApiError(`task_update ID 与订阅的 ${expectedTaskId} 不一致`, 200) + } + if (backendTaskType(value.task_type) !== expectedBackendType(expectation.type)) { + throw new GenerationApiError(`task_update 类型与 ${expectation.type} 不匹配`, 200) + } + if (dtoPositiveInteger(value.project_id, 'project_id') !== expectedProjectId) { + throw new GenerationApiError('task_update 不属于当前项目', 200) + } + if (dtoPositiveInteger(value.user_id, 'user_id') !== expectedUserId) { + throw new GenerationApiError('task_update 不属于当前用户', 200) + } + const inputPayload = dtoNullableRecord(value.input_payload, 'input_payload') + validateInputPayload(inputPayload, expectation) + const status = taskStatus(value.status) + const result = dtoNullableRecord(value.result, 'result') + const error = dtoNullableString(value.error_message, 'error_message') + validateStatusError(status, error) + return { + taskId: String(taskId), + type: expectation.type, + status, + result: mapResult(result, status, expectation), + error, + } +} + +/** + * 创建 Generation 实体适配器。 + * + * `userId` 与 HTTP/SSE transport 都由宿主注入，因此模块既不持有登录态，也不直接 + * 依赖具体 fetch/SSE 实现。三个前端阶段在这里收口为后端的两类 GenerationTask。 + */ +export function createGenerationApis(config: GenerationApiConfig): GenerationApis { + const userId = inputPositiveInteger(config.userId, 'userId') + const { request, stream } = config.transport + const expectations = new Map() + + 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 + } + + const apis: GenerationApis = { + 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() !== '') + const expectation = { type: input.type, actionType: input.actionType } as Extract< + GenerationExpectation, + { type: T['type'] } + > + const generation = await post('/generation/action', projectId, expectation, { + 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, + }) + expectations.set(generation.id, expectation) + return generation as Generation + } + + const expectation = { type: 'character_template' } as const + const generation = await post('/generation/image', projectId, expectation, { + 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, + }) + expectations.set(generation.id, expectation) + return generation as Generation + }, + + async get( + projectId: string, + id: string, + expectation?: GenerationExpectation, + ): 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' }, + ) + const raw = await readData(response) + const resolvedExpectation = expectation ?? inferExpectation(parseTaskDto(raw)) + const generation = mapTask(raw, numericProjectId, userId, resolvedExpectation, numericTaskId) + expectations.set(generation.id, resolvedExpectation) + return generation + }, + + subscribe( + projectId: string, + id: string, + expectationOrOnEvent: GenerationExpectation | ((event: GenerationEvent) => void), + onEventOrError?: ((event: GenerationEvent) => void) | ((error: Error) => void), + maybeOnError?: (error: Error) => void, + ): () => void { + const numericProjectId = inputPositiveInteger(projectId, 'projectId') + const numericTaskId = inputPositiveInteger(id, 'taskId') + const expectation = + typeof expectationOrOnEvent === 'function' ? expectations.get(id) : expectationOrOnEvent + if (!expectation) { + throw new GenerationApiError('订阅前必须先创建或查询生成任务') + } + const onEvent = + typeof expectationOrOnEvent === 'function' + ? expectationOrOnEvent + : (onEventOrError as (event: GenerationEvent) => void) + const onError = + typeof expectationOrOnEvent === 'function' + ? () => undefined + : (maybeOnError ?? (() => undefined)) + return stream( + endpoint( + config.baseUrl, + `/generation/tasks/${numericTaskId}/stream?project_id=${numericProjectId}`, + ), + { + eventName: 'task_update', + onEvent(data) { + const event = mapEvent( + parseEventData(data), + numericProjectId, + userId, + numericTaskId, + expectation, + ) + onEvent(event as GenerationEvent) + return event.status === 'completed' || event.status === 'failed' + }, + onError, + }, + ) + }, + } + + return apis +} diff --git a/frontend/src/entities/generation/index.ts b/frontend/src/entities/generation/index.ts index d6841b6f..09f08041 100644 --- a/frontend/src/entities/generation/index.ts +++ b/frontend/src/entities/generation/index.ts @@ -24,6 +24,11 @@ export type TaskStatus = 'pending' | 'running' | 'completed' | 'failed' */ export type GenerationType = 'character_template' | 'first_frame' | 'complete_animation' +export type GenerationExpectation = + | { type: 'character_template' } + | { type: 'first_frame'; actionType: ActionType } + | { type: 'complete_animation'; actionType: ActionType } + interface GenerationInputBase { projectId: string /** 可选参考媒体；没有参考图时传空数组。 */ @@ -141,10 +146,25 @@ export interface GenerationApis { * projectId 不能从 id 推导，后端查询接口要求两者同时传入。 */ get(projectId: Generation['projectId'], id: Generation['id']): Promise + get( + projectId: Generation['projectId'], + id: Generation['id'], + expectation: Extract, + ): Promise> /** 订阅状态变化，返回取消订阅函数。 */ subscribe( projectId: Generation['projectId'], id: Generation['id'], onEvent: (event: GenerationEvent) => void, ): () => void + subscribe( + projectId: Generation['projectId'], + id: Generation['id'], + expectation: Extract, + onEvent: (event: GenerationEvent) => void, + onError: (error: Error) => void, + ): () => void } + +export { createGenerationApis, GenerationApiError } from './api' +export type { GenerationApiConfig, GenerationTransport } from './api' diff --git a/frontend/src/entities/index.ts b/frontend/src/entities/index.ts index 58e769e1..4b010474 100644 --- a/frontend/src/entities/index.ts +++ b/frontend/src/entities/index.ts @@ -32,6 +32,7 @@ export { characterApis } from './character' export type { ActionTemplate, ActionTemplateApis } from './action-template' /* 生成 —— 业务数据，不是「调用生成能力」 */ +export { createGenerationApis, GenerationApiError } from './generation/api' export type { CharacterTemplateGenerationInput, CharacterTemplateGenerationResult, @@ -43,12 +44,14 @@ export type { Generation, GenerationApis, GenerationEvent, + GenerationExpectation, GenerationInput, GenerationResult, GenerationResultFor, GenerationType, TaskStatus, } from './generation' +export type { GenerationApiConfig, GenerationTransport } from './generation/api' /* 媒体上传 —— 页面只依赖公开工厂与不透明引用，不处理 multipart 协议。 */ export { createMediaApis } from './media/api' 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() + } +} From 275b97e89a1ebec15bad33af66c2de509dab213c Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:07:19 +0800 Subject: [PATCH 2/5] test(generation): cover adapter protocol boundaries --- frontend/src/entities/generation/api.test.ts | 301 ++++++++++++++++++- frontend/src/shared/api/stream.test.ts | 90 ++++++ 2 files changed, 389 insertions(+), 2 deletions(-) diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts index 767010f0..895095e7 100644 --- a/frontend/src/entities/generation/api.test.ts +++ b/frontend/src/entities/generation/api.test.ts @@ -102,7 +102,11 @@ describe('createGenerationApis', () => { type: 'character_action', action_type: 'idle', frames: [ - { index: 0, image_url: 'https://cdn.test/first-frame.png', duration_ms: null }, + { + index: 0, + image_url: 'https://cdn.test/first-frame.png', + duration_ms: null, + }, ], }, }), @@ -320,7 +324,11 @@ describe('createGenerationApis', () => { expect(generation.type).toBe('complete_animation') expect(onEvent).toHaveBeenCalledWith( - expect.objectContaining({ taskId: '91', type: 'complete_animation', status: 'completed' }), + expect.objectContaining({ + taskId: '91', + type: 'complete_animation', + status: 'completed', + }), ) }) @@ -377,4 +385,293 @@ describe('createGenerationApis', () => { 'completed 任务不应携带 error_message', ) }) + + it.each([ + ['非对象任务', success([]), '生成任务响应不是对象'], + [ + '无效业务码', + new Response(JSON.stringify({ code: '200', message: 'ok', data: taskData() })), + '生成接口响应缺少有效的 code', + ], + [ + '业务失败', + new Response( + JSON.stringify({ + code: 503, + message: 'provider unavailable', + data: null, + }), + ), + 'provider unavailable', + ], + [ + '缺少数据', + new Response(JSON.stringify({ code: 200, message: 'ok', data: null })), + '生成接口成功响应缺少 data', + ], + ['非 JSON 响应', new Response('not-json', { status: 502 }), '无法解析的响应'], + ])('拒绝%s', async (_label, response, message) => { + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(async () => response), + stream: vi.fn(() => vi.fn()), + }, + }) + + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow(message) + }) + + it.each([ + ['任务 id', { id: 0 }, '生成任务 id 无效'], + ['任务类型', { task_type: 'video' }, '生成任务 task_type 无效'], + ['项目归属', { project_id: 43 }, '生成任务未归属请求中的项目 42'], + ['用户归属', { user_id: 8 }, '生成任务未归属当前用户'], + ['请求任务 id', { id: 92 }, '生成任务 ID 与请求的 91 不一致'], + ['输入对象', { input_payload: [] }, '生成任务 input_payload 无效'], + ['结果对象', { result: [] }, '生成任务 result 无效'], + ['错误字段', { error_message: 1 }, '生成任务 error_message 无效'], + ['任务输入', { input_payload: { num_images: 3 } }, 'num_images 必须为 4'], + [ + '图片结果类型', + { result: { type: 'video', image_urls: ['a', 'b', 'c', 'd'] } }, + '角色图片结果 type 无效', + ], + ['图片数量', { result: { type: 'character_image', image_urls: ['a'] } }, '必须包含 4 个候选'], + ['完成结果', { result: null }, '完成任务缺少 result'], + ])('校验%s', async (_label, overrides, message) => { + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(async () => success(taskData(overrides))), + stream: vi.fn(() => vi.fn()), + }, + }) + + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow(message) + }) + + it.each([ + [ + '非动作结果', + { type: 'video', action_type: 'walk', frames: actionFrames(32) }, + '完整动画结果 type 无效', + ], + [ + '未知动作', + { + type: 'character_action', + action_type: 'dance', + frames: actionFrames(32), + }, + '完整动画结果 action_type 无效', + ], + [ + '空帧', + { type: 'character_action', action_type: 'walk', frames: [] }, + '完整动画结果 frames 无效', + ], + [ + '非对象帧', + { type: 'character_action', action_type: 'walk', frames: [null] }, + '动作帧不是对象', + ], + [ + '无效索引', + { + type: 'character_action', + action_type: 'walk', + frames: [{ index: -1, image_url: 'a', duration_ms: 1 }], + }, + '动作帧 index 无效', + ], + [ + '重复索引', + { + type: 'character_action', + action_type: 'walk', + frames: [ + { index: 0, image_url: 'a', duration_ms: 1 }, + { index: 0, image_url: 'b', duration_ms: 1 }, + ], + }, + '动作帧 index 重复', + ], + [ + '空地址', + { + type: 'character_action', + action_type: 'walk', + frames: [{ index: 0, image_url: '', duration_ms: 1 }], + }, + '动作帧 image_url 无效', + ], + [ + '无效时长', + { + type: 'character_action', + action_type: 'walk', + frames: [{ index: 0, image_url: 'a', duration_ms: -1 }], + }, + '动作帧 duration_ms 无效', + ], + ])('拒绝%s', async (_label, result, message) => { + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(async () => + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 32, action_type: 'walk' }, + result, + }), + ), + ), + stream: vi.fn(() => vi.fn()), + }, + }) + + await expect( + apis.get('42', '91', { type: 'complete_animation', actionType: 'walk' }), + ).rejects.toThrow(message) + }) + + it('映射运行中和失败任务，并拒绝缺失的失败原因', async () => { + const request = vi + .fn() + .mockResolvedValueOnce(success(taskData({ status: 'running', result: null }))) + .mockResolvedValueOnce( + success( + taskData({ + status: 'failed', + result: null, + error_message: 'provider failed', + }), + ), + ) + .mockResolvedValueOnce(success(taskData({ status: 'failed', result: null }))) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91', { type: 'character_template' })).resolves.toMatchObject({ + status: 'running', + result: null, + }) + await expect(apis.get('42', '91', { type: 'character_template' })).resolves.toMatchObject({ + status: 'failed', + error: 'provider failed', + }) + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow( + '失败任务缺少 error_message', + ) + }) + + it('拒绝无缓存阶段的简写订阅并转发显式订阅错误', () => { + let onStreamError: ((error: Error) => void) | undefined + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(), + stream: vi.fn((_url, options) => { + onStreamError = options.onError + return vi.fn() + }), + }, + }) + expect(() => apis.subscribe('42', '91', vi.fn())).toThrow('订阅前必须先创建或查询生成任务') + + const onError = vi.fn() + apis.subscribe('42', '91', { type: 'character_template' }, vi.fn(), onError) + const error = new Error('stream failed') + onStreamError?.(error) + expect(onError).toHaveBeenCalledWith(error) + }) + + it('校验调用参数、原始响应和任务类型边界', async () => { + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi + .fn() + .mockResolvedValueOnce(new Response(JSON.stringify([]))) + .mockResolvedValueOnce(success(taskData({ task_type: 'character_action' }))), + stream: vi.fn(() => vi.fn()), + }, + }) + + await expect(apis.get('invalid', '91', { type: 'character_template' })).rejects.toThrow( + 'projectId 必须是正整数', + ) + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow( + '生成接口响应不是对象', + ) + await expect(apis.get('42', '91', { type: 'character_template' })).rejects.toThrow( + '生成任务类型与 character_template 不匹配', + ) + }) + + it('推断动作首帧阶段，并拒绝无法推断的动作输入', async () => { + const firstFrame = 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.png', duration_ms: null }], + }, + }) + const request = vi + .fn() + .mockResolvedValueOnce(success(firstFrame)) + .mockResolvedValueOnce( + success(taskData({ task_type: 'character_action', input_payload: null })), + ) + .mockResolvedValueOnce( + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 2, action_type: 'walk' }, + }), + ), + ) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + }) + + await expect(apis.get('42', '91')).resolves.toMatchObject({ type: 'first_frame' }) + await expect(apis.get('42', '91')).rejects.toThrow('动作任务缺少 input_payload') + await expect(apis.get('42', '91')).rejects.toThrow( + 'input_payload.num_frames 无法映射到前端阶段', + ) + }) + + it.each([ + ['not-json', 'task_update 不是有效 JSON'], + [JSON.stringify({ ...taskData(), id: 92 }), 'task_update ID 与订阅的 91 不一致'], + [ + JSON.stringify({ ...taskData(), task_type: 'character_action' }), + 'task_update 类型与 character_template 不匹配', + ], + [JSON.stringify({ ...taskData(), project_id: 43 }), 'task_update 不属于当前项目'], + [JSON.stringify({ ...taskData(), user_id: 8 }), 'task_update 不属于当前用户'], + ])('拒绝非法订阅事件', (payload, message) => { + let onStreamEvent: ((data: string) => boolean) | undefined + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(), + stream: vi.fn((_url, options) => { + onStreamEvent = options.onEvent + return vi.fn() + }), + }, + }) + apis.subscribe('42', '91', { type: 'character_template' }, vi.fn(), vi.fn()) + + expect(() => onStreamEvent?.(payload)).toThrow(message) + }) }) diff --git a/frontend/src/shared/api/stream.test.ts b/frontend/src/shared/api/stream.test.ts index e4ab9143..cc836680 100644 --- a/frontend/src/shared/api/stream.test.ts +++ b/frontend/src/shared/api/stream.test.ts @@ -164,4 +164,94 @@ describe('createEventStreamSubscriber', () => { ) expect(fetchFn).toHaveBeenCalledOnce() }) + + it.each([ + [new Response(null, { status: 503 }), 'SSE 请求失败（HTTP 503）'], + [new Response('plain text'), 'SSE 响应类型无效'], + [ + new Response(null, { headers: { 'content-type': 'text/event-stream' } }), + 'SSE 响应缺少消息流', + ], + ])('报告不可恢复的响应协议错误', async (response, message) => { + const onError = vi.fn() + const subscriber = createEventStreamSubscriber({ + fetchFn: vi.fn(async () => response), + getAccessToken: () => null, + }) + + subscriber('https://api.test/stream', { + eventName: 'task_update', + onEvent: () => false, + onError, + }) + + await vi.waitFor(() => + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message })), + ) + }) + + it('刷新失败后报告 401，且不会重复刷新', async () => { + const recoverUnauthorized = vi.fn(async () => false) + const onError = vi.fn() + const subscriber = createEventStreamSubscriber({ + fetchFn: vi.fn(async () => new Response(null, { status: 401 })), + getAccessToken: () => 'expired', + recoverUnauthorized, + }) + + subscriber('https://api.test/stream', { + eventName: 'task_update', + onEvent: () => false, + onError, + }) + + await vi.waitFor(() => expect(onError).toHaveBeenCalled()) + expect(recoverUnauthorized).toHaveBeenCalledOnce() + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ status: 401 })) + }) + + it('忽略注释和其他事件，并在流结束后重连', async () => { + const encoder = new TextEncoder() + const first = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(': keepalive\r\nevent: progress\r\ndata: 10\r\n\r\n')) + controller.enqueue(encoder.encode('event: task_update\ndata: {"status":')) + controller.enqueue(encoder.encode('"running"}\n\n')) + controller.close() + }, + }), + { headers: { 'content-type': 'text/event-stream; charset=utf-8' } }, + ) + const fetchFn = vi + .fn() + .mockResolvedValueOnce(first) + .mockResolvedValueOnce(eventStreamResponse('{"status":"completed"}')) + const onEvent = vi.fn((data: string) => data.includes('completed')) + const onError = vi.fn() + const subscriber = createEventStreamSubscriber({ + fetchFn, + getAccessToken: () => null, + reconnectDelayMs: 0, + }) + + await new Promise((resolve) => { + subscriber('https://api.test/stream', { + eventName: 'task_update', + onEvent(data) { + const terminal = onEvent(data) + if (terminal) resolve() + return terminal + }, + onError, + }) + }) + + expect(onEvent).toHaveBeenNthCalledWith(1, '{"status":"running"}') + expect(onEvent).toHaveBeenNthCalledWith(2, '{"status":"completed"}') + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: 'SSE 连接中断，正在自动重连' }), + ) + expect(fetchFn).toHaveBeenCalledTimes(2) + }) }) From 68421f40d4392e203d90867481db9fc0a15078fb Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:24:54 +0800 Subject: [PATCH 3/5] fix(generation): reconcile SSE event contracts --- frontend/src/entities/generation/api.test.ts | 146 ++++++++++++++++-- frontend/src/entities/generation/api.ts | 127 +++++++++++++-- frontend/src/entities/generation/index.ts | 9 +- frontend/src/features/export/index.test.ts | 8 +- frontend/src/features/export/index.ts | 6 +- .../workflow-controller/controller.test.ts | 2 + .../src/pages/workflow-editor/index.test.tsx | 4 +- .../src/pages/workflow-editor/runtime.test.ts | 4 +- frontend/src/shared/api/stream.test.ts | 22 +++ frontend/src/shared/api/stream.ts | 15 +- 10 files changed, 294 insertions(+), 49 deletions(-) diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts index 895095e7..a2adc8f7 100644 --- a/frontend/src/entities/generation/api.test.ts +++ b/frontend/src/entities/generation/api.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { createGenerationApis, GenerationApiError } from '@/entities' +import { EventStreamError, type EventStreamOptions } from '@/shared/api/stream' import type { MediaReference } from '../media' @@ -188,6 +189,7 @@ describe('createGenerationApis', () => { expect(generation.result).toEqual({ type: 'complete_animation', frames: Array.from({ length: 32 }, (_, index) => ({ + index, url: `https://cdn.test/frame-${index + 1}.png`, durationMs: index % 2 === 0 ? 100 : null, })), @@ -225,13 +227,7 @@ describe('createGenerationApis', () => { it('订阅 task_update，映射终态并把终态关闭信号交给流传输层', () => { let subscribedUrl = '' - let streamOptions: - | { - eventName: string - onEvent(data: string): boolean - onError(error: Error): void - } - | undefined + let streamOptions: EventStreamOptions | undefined const cancel = vi.fn() const stream = vi.fn((url: string, options: NonNullable) => { subscribedUrl = url @@ -268,10 +264,11 @@ describe('createGenerationApis', () => { }, error_message: null, }), + 'task_update', ) expect(subscribedUrl).toBe('https://api.test/generation/tasks/91/stream?project_id=42') - expect(streamOptions?.eventName).toBe('task_update') + expect(streamOptions?.eventName).toEqual(['task_update', 'progress', 'completed', 'failed']) expect(isTerminal).toBe(true) expect(onEvent).toHaveBeenCalledWith({ taskId: '91', @@ -280,6 +277,7 @@ describe('createGenerationApis', () => { result: { type: 'complete_animation', frames: Array.from({ length: 32 }, (_, index) => ({ + index, url: `https://cdn.test/frame-${index + 1}.png`, durationMs: index % 2 === 0 ? 100 : null, })), @@ -292,13 +290,7 @@ describe('createGenerationApis', () => { }) it('从查询结果推断前端阶段，并允许现有三参数订阅继续使用', async () => { - let streamOptions: - | { - eventName: string - onEvent(data: string): boolean - onError(error: Error): void - } - | undefined + let streamOptions: EventStreamOptions | undefined const task = taskData({ task_type: 'character_action', input_payload: { num_frames: 32, action_type: 'walk' }, @@ -320,7 +312,7 @@ describe('createGenerationApis', () => { const generation = await apis.get('42', '91') const onEvent = vi.fn() apis.subscribe('42', '91', onEvent) - streamOptions?.onEvent(JSON.stringify(task)) + streamOptions?.onEvent(JSON.stringify(task), 'task_update') expect(generation.type).toBe('complete_animation') expect(onEvent).toHaveBeenCalledWith( @@ -674,4 +666,126 @@ describe('createGenerationApis', () => { expect(() => onStreamEvent?.(payload)).toThrow(message) }) + + it('接受只含 task_id 的精简终态事件并保留动作帧元数据', () => { + let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined + const onEvent = vi.fn() + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(), + stream: vi.fn((_url, options) => { + onStreamEvent = options.onEvent + return vi.fn() + }), + }, + }) + apis.subscribe('42', '91', { type: 'complete_animation', actionType: 'walk' }, onEvent, vi.fn()) + + const terminal = onStreamEvent?.( + JSON.stringify({ + task_id: 91, + task_type: 'character_action', + status: 'completed', + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(32), + }, + }), + 'task_update', + ) + + expect(terminal).toBe(true) + expect(onEvent).toHaveBeenCalledWith( + expect.objectContaining({ + taskId: '91', + status: 'completed', + result: expect.objectContaining({ + frames: expect.arrayContaining([ + { index: 0, url: 'https://cdn.test/frame-1.png', durationMs: 100 }, + ]), + }), + }), + ) + }) + + it('拒绝同时存在但不一致的 task_id 与 id', () => { + let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(), + stream: vi.fn((_url, options) => { + onStreamEvent = options.onEvent + return vi.fn() + }), + }, + }) + apis.subscribe('42', '91', { type: 'character_template' }, vi.fn(), vi.fn()) + + expect(() => + onStreamEvent?.(JSON.stringify({ ...taskData(), task_id: 91, id: 92 }), 'task_update'), + ).toThrow('task_update 的 task_id 与 id 不一致') + }) + + it('根据 completed 事件名补全精简 payload 的终态', () => { + let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined + const onEvent = vi.fn() + const apis = createGenerationApis({ + userId: 7, + transport: { + request: vi.fn(), + stream: vi.fn((_url, options) => { + onStreamEvent = options.onEvent + return vi.fn() + }), + }, + }) + apis.subscribe('42', '91', { type: 'character_template' }, onEvent, vi.fn()) + + expect( + onStreamEvent?.( + JSON.stringify({ + task_id: 91, + task_type: 'character_image', + result: taskData().result, + }), + 'completed', + ), + ).toBe(true) + expect(onEvent).toHaveBeenCalledWith(expect.objectContaining({ status: 'completed' })) + }) + + it('SSE 路由缺失时轮询任务查询直到终态', async () => { + const request = vi + .fn() + .mockResolvedValueOnce(success(taskData({ status: 'running', result: null }))) + .mockResolvedValueOnce(success(taskData())) + const onEvent = vi.fn() + const apis = createGenerationApis({ + userId: 7, + pollIntervalMs: 1, + transport: { + request, + stream: vi.fn((_url, options) => { + queueMicrotask(() => + options.onError( + new EventStreamError('SSE 请求失败（HTTP 404）', false, undefined, 404), + ), + ) + return vi.fn() + }), + }, + }) + + apis.subscribe('42', '91', { type: 'character_template' }, onEvent, vi.fn()) + + await vi.waitFor(() => + expect(onEvent).toHaveBeenLastCalledWith( + expect.objectContaining({ taskId: '91', status: 'completed' }), + ), + ) + expect(request).toHaveBeenCalledTimes(2) + }) }) diff --git a/frontend/src/entities/generation/api.ts b/frontend/src/entities/generation/api.ts index 2f31b542..48996216 100644 --- a/frontend/src/entities/generation/api.ts +++ b/frontend/src/entities/generation/api.ts @@ -1,4 +1,4 @@ -import type { EventStreamSubscriber } from '@/shared/api/stream' +import { EventStreamError, type EventStreamSubscriber } from '@/shared/api/stream' import type { CompleteAnimationGenerationInput, @@ -27,6 +27,8 @@ export interface GenerationApiConfig { /** 当前用户由认证宿主提供，适配器不猜测也不写死身份。 */ userId: string | number transport: GenerationTransport + /** SSE 路由不存在时，任务查询兜底的间隔。 */ + pollIntervalMs?: number } interface ResponseEnvelope { @@ -250,7 +252,7 @@ function mapActionResult( } return { type: 'complete_animation', - frames: orderedFrames.map(({ url, durationMs }) => ({ url, durationMs })), + frames: orderedFrames, } } @@ -383,33 +385,80 @@ function parseEventData(data: string): unknown { } } +function eventTaskId(value: Record): number { + const taskId = value.task_id === undefined ? null : dtoPositiveInteger(value.task_id, 'task_id') + const id = value.id === undefined ? null : dtoPositiveInteger(value.id, 'id') + if (taskId !== null && id !== null && taskId !== id) { + throw new GenerationApiError('task_update 的 task_id 与 id 不一致', 200) + } + if (taskId === null && id === null) return dtoPositiveInteger(undefined, 'task_id') + return taskId ?? id! +} + +function eventStatus(value: Record, eventName: string): TaskStatus { + const impliedStatus = + eventName === 'completed' ? 'completed' : eventName === 'failed' ? 'failed' : null + if (value.status === undefined) { + if (impliedStatus) return impliedStatus + if (eventName === 'progress') return 'running' + } + const status = taskStatus(value.status) + if (impliedStatus && status !== impliedStatus) { + throw new GenerationApiError(`SSE ${eventName} 事件与 status=${status} 不一致`, 200) + } + return status +} + +function waitForPoll(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 }) + }) +} + function mapEvent( value: unknown, expectedProjectId: number, expectedUserId: number, expectedTaskId: number, expectation: Extract, + eventName: string, ): GenerationEvent { if (!isRecord(value)) throw new GenerationApiError('task_update 不是对象', 200) - // 后端 PR #34 推送完整 GenerationTask，标识字段是数字 id；前端统一转为字符串 taskId。 - const taskId = dtoPositiveInteger(value.id, 'id') + const taskId = eventTaskId(value) if (taskId !== expectedTaskId) { throw new GenerationApiError(`task_update ID 与订阅的 ${expectedTaskId} 不一致`, 200) } if (backendTaskType(value.task_type) !== expectedBackendType(expectation.type)) { throw new GenerationApiError(`task_update 类型与 ${expectation.type} 不匹配`, 200) } - if (dtoPositiveInteger(value.project_id, 'project_id') !== expectedProjectId) { + if ( + value.project_id !== undefined && + dtoPositiveInteger(value.project_id, 'project_id') !== expectedProjectId + ) { throw new GenerationApiError('task_update 不属于当前项目', 200) } - if (dtoPositiveInteger(value.user_id, 'user_id') !== expectedUserId) { + if ( + value.user_id !== undefined && + dtoPositiveInteger(value.user_id, 'user_id') !== expectedUserId + ) { throw new GenerationApiError('task_update 不属于当前用户', 200) } - const inputPayload = dtoNullableRecord(value.input_payload, 'input_payload') - validateInputPayload(inputPayload, expectation) - const status = taskStatus(value.status) - const result = dtoNullableRecord(value.result, 'result') - const error = dtoNullableString(value.error_message, 'error_message') + if (value.input_payload !== undefined) { + validateInputPayload(dtoNullableRecord(value.input_payload, 'input_payload'), expectation) + } + const status = eventStatus(value, eventName) + const result = value.result === undefined ? null : dtoNullableRecord(value.result, 'result') + const error = + value.error_message === undefined + ? null + : dtoNullableString(value.error_message, 'error_message') validateStatusError(status, error) return { taskId: String(taskId), @@ -429,6 +478,10 @@ function mapEvent( export function createGenerationApis(config: GenerationApiConfig): GenerationApis { const userId = inputPositiveInteger(config.userId, 'userId') const { request, stream } = config.transport + const pollIntervalMs = config.pollIntervalMs ?? 1_000 + if (!Number.isFinite(pollIntervalMs) || pollIntervalMs < 0) { + throw new GenerationApiError('pollIntervalMs 必须是非负数') + } const expectations = new Map() async function post( @@ -529,27 +582,71 @@ export function createGenerationApis(config: GenerationApiConfig): GenerationApi typeof expectationOrOnEvent === 'function' ? () => undefined : (maybeOnError ?? (() => undefined)) - return stream( + const pollingController = new AbortController() + let polling = false + let stopStream: () => void = () => undefined + + const pollUntilTerminal = async () => { + if (polling) return + polling = true + while (!pollingController.signal.aborted) { + try { + const generation = await apis.get(projectId, id, expectation) + if (pollingController.signal.aborted) return + onEvent({ + taskId: generation.id, + type: generation.type, + status: generation.status, + result: generation.result, + error: generation.error, + } as GenerationEvent) + if (generation.status === 'completed' || generation.status === 'failed') return + } catch (cause) { + if (!pollingController.signal.aborted) { + onError(cause instanceof Error ? cause : new GenerationApiError('任务轮询失败')) + } + return + } + await waitForPoll(pollIntervalMs, pollingController.signal) + } + } + + stopStream = stream( endpoint( config.baseUrl, `/generation/tasks/${numericTaskId}/stream?project_id=${numericProjectId}`, ), { - eventName: 'task_update', - onEvent(data) { + eventName: ['task_update', 'progress', 'completed', 'failed'], + onEvent(data, eventName) { const event = mapEvent( parseEventData(data), numericProjectId, userId, numericTaskId, expectation, + eventName, ) onEvent(event as GenerationEvent) return event.status === 'completed' || event.status === 'failed' }, - onError, + onError(error) { + if ( + error instanceof EventStreamError && + (error.status === 404 || error.status === 405 || error.status === 501) + ) { + stopStream() + void pollUntilTerminal() + return + } + onError(error) + }, }, ) + return () => { + stopStream() + pollingController.abort() + } }, } diff --git a/frontend/src/entities/generation/index.ts b/frontend/src/entities/generation/index.ts index 09f08041..b2fe9b61 100644 --- a/frontend/src/entities/generation/index.ts +++ b/frontend/src/entities/generation/index.ts @@ -76,6 +76,12 @@ export interface GeneratedImage { url: string } +/** 后端动作帧字段完整映射，不能用数组位置覆盖服务端 index。 */ +export interface GeneratedFrame extends GeneratedImage { + index: number + durationMs: number | null +} + /** 结果按 type 分别定义，不共用一个 urls 数组。 */ export interface CharacterTemplateGenerationResult { type: 'character_template' @@ -87,10 +93,9 @@ export interface FirstFrameGenerationResult { image: GeneratedImage } -/** 帧顺序由数组位置表达。 */ export interface CompleteAnimationGenerationResult { type: 'complete_animation' - frames: readonly GeneratedImage[] + frames: readonly GeneratedFrame[] } export type GenerationResult = diff --git a/frontend/src/features/export/index.test.ts b/frontend/src/features/export/index.test.ts index c07157e1..c4f85e95 100644 --- a/frontend/src/features/export/index.test.ts +++ b/frontend/src/features/export/index.test.ts @@ -54,8 +54,8 @@ describe('Character asset publisher', () => { fps: 12, frameCount: 2, frames: [ - { index: 0, imageUrl: 'https://assets.windup.test/walk-01.png', durationMs: null }, - { index: 1, imageUrl: 'https://assets.windup.test/walk-02.png', durationMs: null }, + { index: 0, imageUrl: 'https://assets.windup.test/walk-01.png', durationMs: 125 }, + { index: 1, imageUrl: 'https://assets.windup.test/walk-02.png', durationMs: 80 }, ], }, ]) @@ -246,8 +246,8 @@ function completeAnimationFixture(): Generation<'complete_animation'> { result: { type: 'complete_animation', frames: [ - { url: 'https://assets.windup.test/walk-01.png' }, - { url: 'https://assets.windup.test/walk-02.png' }, + { index: 0, url: 'https://assets.windup.test/walk-01.png', durationMs: 125 }, + { index: 1, url: 'https://assets.windup.test/walk-02.png', durationMs: 80 }, ], }, } diff --git a/frontend/src/features/export/index.ts b/frontend/src/features/export/index.ts index 706e80b9..5d8618f5 100644 --- a/frontend/src/features/export/index.ts +++ b/frontend/src/features/export/index.ts @@ -71,10 +71,10 @@ export function createCharacterAssetPublisher( loop: firstFrameNode.input.type === 'idle' || firstFrameNode.input.type === 'walk', fps: firstFrameNode.input.fps, frameCount: generation.result.frames.length, - frames: generation.result.frames.map((frame, index) => ({ - index, + frames: generation.result.frames.map((frame) => ({ + index: frame.index, imageUrl: frame.url, - durationMs: null, + durationMs: frame.durationMs, })), } const targetOutfit = character.outfits[outfitIndex]! diff --git a/frontend/src/features/workflow-controller/controller.test.ts b/frontend/src/features/workflow-controller/controller.test.ts index 54352b97..433605db 100644 --- a/frontend/src/features/workflow-controller/controller.test.ts +++ b/frontend/src/features/workflow-controller/controller.test.ts @@ -239,7 +239,9 @@ function completedAnimationEvent(taskId = 'task-2'): GenerationEvent { result: { type: 'complete_animation', frames: Array.from({ length: 32 }, (_, index) => ({ + index, url: `https://img/frame-${index}.png`, + durationMs: index % 2 === 0 ? 100 : null, })), }, error: null, diff --git a/frontend/src/pages/workflow-editor/index.test.tsx b/frontend/src/pages/workflow-editor/index.test.tsx index 40c5f83e..7b9b9e96 100644 --- a/frontend/src/pages/workflow-editor/index.test.tsx +++ b/frontend/src/pages/workflow-editor/index.test.tsx @@ -943,8 +943,8 @@ function completeAnimationGeneration(): Generation<'complete_animation'> { result: { type: 'complete_animation', frames: [ - { url: 'https://assets.windup.test/walk-01.png' }, - { url: 'https://assets.windup.test/walk-02.png' }, + { index: 0, url: 'https://assets.windup.test/walk-01.png', durationMs: 100 }, + { index: 1, url: 'https://assets.windup.test/walk-02.png', durationMs: null }, ], }, error: null, diff --git a/frontend/src/pages/workflow-editor/runtime.test.ts b/frontend/src/pages/workflow-editor/runtime.test.ts index 1a1187b0..c3a009d9 100644 --- a/frontend/src/pages/workflow-editor/runtime.test.ts +++ b/frontend/src/pages/workflow-editor/runtime.test.ts @@ -364,8 +364,8 @@ function completeAnimationFixture(): Generation<'complete_animation'> { result: { type: 'complete_animation', frames: [ - { url: 'https://assets.windup.test/walk-01.png' }, - { url: 'https://assets.windup.test/walk-02.png' }, + { index: 0, url: 'https://assets.windup.test/walk-01.png', durationMs: 100 }, + { index: 1, url: 'https://assets.windup.test/walk-02.png', durationMs: null }, ], }, } diff --git a/frontend/src/shared/api/stream.test.ts b/frontend/src/shared/api/stream.test.ts index cc836680..7b6499a1 100644 --- a/frontend/src/shared/api/stream.test.ts +++ b/frontend/src/shared/api/stream.test.ts @@ -254,4 +254,26 @@ describe('createEventStreamSubscriber', () => { ) expect(fetchFn).toHaveBeenCalledTimes(2) }) + + it('把匹配到的实际事件名交给业务解析器', async () => { + const onEvent = vi.fn((_data: string, _eventName: string) => true) + const subscriber = createEventStreamSubscriber({ + fetchFn: vi.fn(async () => eventStreamResponse('{"task_id":91}', 'completed')), + getAccessToken: () => null, + }) + + await new Promise((resolve, reject) => { + subscriber('https://api.test/stream', { + eventName: ['task_update', 'progress', 'completed', 'failed'], + onEvent(data, eventName) { + onEvent(data, eventName) + resolve() + return true + }, + onError: reject, + }) + }) + + expect(onEvent).toHaveBeenCalledWith('{"task_id":91}', 'completed') + }) }) diff --git a/frontend/src/shared/api/stream.ts b/frontend/src/shared/api/stream.ts index 33a78d8d..56f93a29 100644 --- a/frontend/src/shared/api/stream.ts +++ b/frontend/src/shared/api/stream.ts @@ -1,10 +1,10 @@ /** 业务无关的、可鉴权的 SSE 订阅边界。 */ export interface EventStreamOptions { - /** 只监听业务指定的命名事件，例如 task_update。 */ - eventName: string + /** 只监听业务指定的命名事件，例如 task_update 或 completed。 */ + eventName: string | readonly string[] /** 返回 true 表示 payload 是终态，传输层随后关闭连接。 */ - onEvent(data: string): boolean + onEvent(data: string, eventName: string): boolean /** 包含连接中断、非法响应和业务解析器抛出的错误。 */ onError(error: Error): void } @@ -73,8 +73,13 @@ async function readEventStream(response: Response, options: EventStreamOptions): const deliver = async (block: string): Promise => { const record = parseRecord(block) - if (record?.event !== options.eventName) return false - if (!options.onEvent(record.data)) return false + if (!record) return false + const matches = + typeof options.eventName === 'string' + ? record.event === options.eventName + : options.eventName.includes(record.event) + if (!matches) return false + if (!options.onEvent(record.data, record.event)) return false await reader.cancel() return true } From b85e4ef2257ec18e9e697f36e17ea1d8f15a383b Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:55:20 +0800 Subject: [PATCH 4/5] fix(generation): derive ownership from access token --- .../app/src/windup_app/web/api/generation.py | 89 ++++++++--- backend/tests/test_generation_api.py | 145 ++++++++++++++++++ frontend/src/entities/generation/api.ts | 6 +- 3 files changed, 212 insertions(+), 28 deletions(-) create mode 100644 backend/tests/test_generation_api.py diff --git a/backend/packages/app/src/windup_app/web/api/generation.py b/backend/packages/app/src/windup_app/web/api/generation.py index a3f80ca3..336f2c5d 100644 --- a/backend/packages/app/src/windup_app/web/api/generation.py +++ b/backend/packages/app/src/windup_app/web/api/generation.py @@ -29,10 +29,12 @@ from windup_common.result import Response from windup_framework.db import get_session +from windup_app.server.character.model import Character from windup_app.server.orchestrator.model import ( ActionType, GenerationTask, ) +from windup_app.server.project.model import Project logger = logging.getLogger("windup.generation.api") @@ -54,23 +56,34 @@ class _EventBus: """任务进度内存发布-订阅。""" def __init__(self) -> None: - self._queues: dict[str, list[asyncio.Queue]] = defaultdict(list) + self._queues: dict[tuple[int, int], list[asyncio.Queue]] = defaultdict(list) - async def subscribe(self, task_id: int) -> asyncio.Queue: + async def subscribe(self, project_id: int, task_id: int) -> asyncio.Queue: queue: asyncio.Queue = asyncio.Queue() - self._queues[str(task_id)].append(queue) + self._queues[(project_id, task_id)].append(queue) return queue - async def unsubscribe(self, task_id: int, queue: asyncio.Queue) -> None: - key = str(task_id) + async def unsubscribe( + self, + project_id: int, + task_id: int, + queue: asyncio.Queue, + ) -> None: + key = (project_id, task_id) subs = self._queues.get(key) if subs and queue in subs: subs.remove(queue) if not subs: del self._queues[key] - def publish(self, task_id: int, event: str, data: dict) -> None: - for queue in self._queues.get(str(task_id), []): + def publish( + self, + project_id: int, + task_id: int, + event: str, + data: dict, + ) -> None: + for queue in self._queues.get((project_id, task_id), []): queue.put_nowait((event, data)) @@ -86,8 +99,7 @@ def publish(self, task_id: int, event: str, data: dict) -> None: class CharacterImageGenerateRequest(BaseModel): """提交角色图片生成任务。""" - user_id: int = Field(gt=0) - project_id: int | None = None + project_id: int = Field(gt=0) reference_image_url: str | None = None prompt: str = "" negative_prompt: str = "" @@ -99,8 +111,7 @@ class CharacterImageGenerateRequest(BaseModel): class CharacterActionGenerateRequest(BaseModel): """提交角色动作生成任务。""" - user_id: int = Field(gt=0) - project_id: int | None = None + project_id: int = Field(gt=0) character_id: int = Field(gt=0) action_type: ActionType custom_prompt: str | None = None @@ -146,15 +157,32 @@ def _task_to_out(task: GenerationTask) -> GenerationTaskOut: # ══════════════════════════════════════════════════════════════════════════════ -def _validate_project_size(session: Session, project_id: int | None, width: int, height: int) -> None: - """校验输入尺寸与项目约束是否一致;不一致则抛异常。""" - if project_id is None: - return - from windup_app.server.project.service import SqlAlchemyProjectService +def _get_project_or_raise( + session: Session, + project_id: int, + user_id: int, +) -> Project: + """校验项目存在且属于 token 对应用户。""" + project = session.get(Project, project_id) + if project is None or project.user_id != user_id: + raise BizException("项目不存在", code=BizCode.NOT_FOUND) + return project + - project = SqlAlchemyProjectService().get_project(session, project_id) - if project is None: - return +def _get_character_or_raise( + session: Session, + character_id: int, + project_id: int, +) -> Character: + """校验角色存在且属于本次生成所指定的项目。""" + character = session.get(Character, character_id) + if character is None or character.project_id != project_id: + raise BizException("角色不存在", code=BizCode.NOT_FOUND) + return character + + +def _validate_project_size(project: Project, width: int, height: int) -> None: + """校验输入尺寸与项目约束是否一致;不一致则抛异常。""" if width != project.sprite_width or height != project.sprite_height: raise BizException( f"输入尺寸 {width}×{height} 与项目约束 {project.sprite_width}×{project.sprite_height} 不一致", @@ -169,7 +197,9 @@ def submit_image_generation( session: Session = Depends(get_session), ) -> Response[GenerationTaskOut]: """提交角色图片生成任务:建 PENDING 记录立即返回,实际图生图后台跑。""" - _validate_project_size(session, body.project_id, body.width, body.height) + user_id = request.state.current_user.id + project = _get_project_or_raise(session, body.project_id, user_id) + _validate_project_size(project, body.width, body.height) # TODO: service.create_image_task + background_tasks.add_task raise BizException("接口待实现", code=BizCode.BAD_REQUEST) @@ -181,6 +211,9 @@ def submit_action_generation( session: Session = Depends(get_session), ) -> Response[GenerationTaskOut]: """提交角色动作生成任务:建 PENDING 记录立即返回,实际生成后台跑。""" + user_id = request.state.current_user.id + _get_project_or_raise(session, body.project_id, user_id) + _get_character_or_raise(session, body.character_id, body.project_id) # TODO: service.create_action_task + background_tasks.add_task raise BizException("接口待实现", code=BizCode.BAD_REQUEST) @@ -189,10 +222,13 @@ def submit_action_generation( def get_task( task_id: int, project_id: int = Query(..., gt=0), + request: Request = None, session: Session = Depends(get_session), ) -> Response[GenerationTaskOut]: """查询生成任务状态与结果。""" - # TODO: service.get_task + user_id = request.state.current_user.id + _get_project_or_raise(session, project_id, user_id) + # TODO: service.get_task，并校验任务属于 project_id raise BizException("接口待实现", code=BizCode.BAD_REQUEST) @@ -212,8 +248,10 @@ async def stream_task( 若客户端订阅时任务已处于终态,立即推送终态事件并关闭连接。 """ - # TODO: 检查任务初始状态,若已终态立即推送 - queue = await event_bus.subscribe(task_id) + user_id = request.state.current_user.id + _get_project_or_raise(session, project_id, user_id) + # TODO: 检查任务属于 project_id 及初始状态,若已终态立即推送 + queue = await event_bus.subscribe(project_id, task_id) logger.debug("SSE 订阅: task_id=%d", task_id) async def _event_generator(): @@ -224,7 +262,8 @@ async def _event_generator(): break try: event, data = await asyncio.wait_for( - queue.get(), timeout=_HEARTBEAT_TIMEOUT, + queue.get(), + timeout=_HEARTBEAT_TIMEOUT, ) payload = json.dumps(data, ensure_ascii=False) yield f"event: {event}\ndata: {payload}\n\n" @@ -234,7 +273,7 @@ async def _event_generator(): except asyncio.TimeoutError: yield ": heartbeat\n\n" finally: - await event_bus.unsubscribe(task_id, queue) + await event_bus.unsubscribe(project_id, task_id, queue) logger.debug("SSE 取消订阅: task_id=%d", task_id) return StreamingResponse( diff --git a/backend/tests/test_generation_api.py b/backend/tests/test_generation_api.py new file mode 100644 index 00000000..67da7bad --- /dev/null +++ b/backend/tests/test_generation_api.py @@ -0,0 +1,145 @@ +"""生成任务 API 的认证与资源归属测试。""" + +import asyncio + +from windup_app.web.api.generation import _EventBus + + +def _create_project(auth_client, name: str = "生成项目") -> dict: + return auth_client.post( + "/projects", + json={ + "project_name": name, + "character_perspective": 1, + "directional_movement": 2, + "sprite_width": 64, + "sprite_height": 64, + }, + ).json()["data"] + + +def _create_character(auth_client, project_id: int) -> dict: + return auth_client.post( + "/characters", + json={ + "project_id": project_id, + "workflow_run_id": 1, + "name": "勇者", + }, + ).json()["data"] + + +def _image_payload(project_id: int, **overrides) -> dict: + payload = { + "project_id": project_id, + "prompt": "像素风勇者", + "width": 64, + "height": 64, + } + payload.update(overrides) + return payload + + +def _action_payload(project_id: int, character_id: int, **overrides) -> dict: + payload = { + "project_id": project_id, + "character_id": character_id, + "action_type": "walk", + } + payload.update(overrides) + return payload + + +def test_image_generation_uses_token_user_without_body_user_id(auth_client): + project = _create_project(auth_client) + + response = auth_client.post( + "/generation/image", + json=_image_payload(project["id"]), + ) + + assert response.json()["code"] == 400 + assert response.json()["message"] == "接口待实现" + + +def test_spoofed_body_user_id_cannot_access_other_users_project( + auth_client, + auth_client_b, +): + project = _create_project(auth_client) + + response = auth_client_b.post( + "/generation/image", + json=_image_payload(project["id"], user_id=1), + ) + + assert response.json()["code"] == 404 + assert response.json()["message"] == "项目不存在" + + +def test_action_generation_uses_token_user_without_body_user_id(auth_client): + project = _create_project(auth_client) + character = _create_character(auth_client, project["id"]) + + response = auth_client.post( + "/generation/action", + json=_action_payload(project["id"], character["id"]), + ) + + assert response.json()["code"] == 400 + assert response.json()["message"] == "接口待实现" + + +def test_action_character_must_belong_to_requested_project(auth_client): + first_project = _create_project(auth_client, "项目一") + second_project = _create_project(auth_client, "项目二") + character = _create_character(auth_client, first_project["id"]) + + response = auth_client.post( + "/generation/action", + json=_action_payload(second_project["id"], character["id"]), + ) + + assert response.json()["code"] == 404 + assert response.json()["message"] == "角色不存在" + + +def test_task_query_checks_project_ownership(auth_client, auth_client_b): + project = _create_project(auth_client) + + response = auth_client_b.get( + "/generation/tasks/1", + params={"project_id": project["id"]}, + ) + + assert response.json()["code"] == 404 + assert response.json()["message"] == "项目不存在" + + +def test_task_stream_checks_project_ownership(auth_client, auth_client_b): + project = _create_project(auth_client) + + response = auth_client_b.get( + "/generation/tasks/1/stream", + params={"project_id": project["id"]}, + ) + + assert response.json()["code"] == 404 + assert response.json()["message"] == "项目不存在" + + +def test_event_bus_isolates_same_task_id_between_projects(): + async def scenario(): + bus = _EventBus() + first_queue = await bus.subscribe(1, 9) + second_queue = await bus.subscribe(2, 9) + + bus.publish(1, 9, "progress", {"status": "running"}) + + assert first_queue.get_nowait() == ( + "progress", + {"status": "running"}, + ) + assert second_queue.empty() + + asyncio.run(scenario()) diff --git a/frontend/src/entities/generation/api.ts b/frontend/src/entities/generation/api.ts index 48996216..5cab6fef 100644 --- a/frontend/src/entities/generation/api.ts +++ b/frontend/src/entities/generation/api.ts @@ -24,7 +24,7 @@ export interface GenerationTransport { export interface GenerationApiConfig { /** API 前缀；空字符串表示同源。 */ baseUrl?: string - /** 当前用户由认证宿主提供，适配器不猜测也不写死身份。 */ + /** 仅用于校验响应归属；请求授权统一由 transport 携带的 token 决定。 */ userId: string | number transport: GenerationTransport /** SSE 路由不存在时，任务查询兜底的间隔。 */ @@ -472,8 +472,8 @@ function mapEvent( /** * 创建 Generation 实体适配器。 * - * `userId` 与 HTTP/SSE transport 都由宿主注入，因此模块既不持有登录态，也不直接 - * 依赖具体 fetch/SSE 实现。三个前端阶段在这里收口为后端的两类 GenerationTask。 + * HTTP/SSE transport 由宿主注入并统一携带 token；`userId` 仅核对响应归属，不会 + * 写入请求或参与后端授权。三个前端阶段在这里收口为后端的两类 GenerationTask。 */ export function createGenerationApis(config: GenerationApiConfig): GenerationApis { const userId = inputPositiveInteger(config.userId, 'userId') From 9fe474e842775be0d1ac7c1104d4e8971e2ca37b Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 11 Aug 2026 16:11:11 +0800 Subject: [PATCH 5/5] refactor(auth): remove client user id contracts --- .../app/src/windup_app/web/api/generation.py | 2 -- .../app/src/windup_app/web/api/project.py | 7 +++-- backend/tests/test_generation_api.py | 6 +++- backend/tests/test_project_api.py | 11 ++++++-- frontend/src/entities/generation/api.test.ts | 25 ----------------- frontend/src/entities/generation/api.ts | 28 ++++--------------- frontend/src/entities/project/index.test.ts | 6 ++-- frontend/src/entities/project/index.ts | 12 ++------ .../src/pages/workflow-editor/index.test.tsx | 1 - .../src/pages/workflow-editor/runtime.test.ts | 1 - frontend/src/test/project-assets-backend.ts | 5 ---- 11 files changed, 26 insertions(+), 78 deletions(-) diff --git a/backend/packages/app/src/windup_app/web/api/generation.py b/backend/packages/app/src/windup_app/web/api/generation.py index 336f2c5d..b03e8afe 100644 --- a/backend/packages/app/src/windup_app/web/api/generation.py +++ b/backend/packages/app/src/windup_app/web/api/generation.py @@ -126,7 +126,6 @@ class GenerationTaskOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int - user_id: int project_id: int | None = None task_type: str status: str @@ -142,7 +141,6 @@ def _task_to_out(task: GenerationTask) -> GenerationTaskOut: result_dict = dataclasses.asdict(task.result) return GenerationTaskOut( id=task.id, - user_id=task.user_id, project_id=task.project_id, task_type=task.task_type.value, status=task.status.value, diff --git a/backend/packages/app/src/windup_app/web/api/project.py b/backend/packages/app/src/windup_app/web/api/project.py index ead71016..7b3a5cc7 100644 --- a/backend/packages/app/src/windup_app/web/api/project.py +++ b/backend/packages/app/src/windup_app/web/api/project.py @@ -39,7 +39,6 @@ class ProjectOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int - user_id: int workflow_id: int | None project_name: str character_perspective: int @@ -64,7 +63,8 @@ def create_project( ): logger.warning( "[WINDUP] 创建拒绝-名称重复 | user_id=%s project_name=%s", - user_id, body.project_name, + user_id, + body.project_name, ) raise BizException("项目名称已存在", code=BizCode.BAD_REQUEST) try: @@ -72,7 +72,8 @@ def create_project( except IntegrityError: logger.warning( "[WINDUP] 创建拒绝-并发冲突 | user_id=%s project_name=%s", - user_id, body.project_name, + user_id, + body.project_name, ) session.rollback() raise BizException("项目名称已存在", code=BizCode.BAD_REQUEST) from None diff --git a/backend/tests/test_generation_api.py b/backend/tests/test_generation_api.py index 67da7bad..24b86d26 100644 --- a/backend/tests/test_generation_api.py +++ b/backend/tests/test_generation_api.py @@ -2,7 +2,7 @@ import asyncio -from windup_app.web.api.generation import _EventBus +from windup_app.web.api.generation import GenerationTaskOut, _EventBus def _create_project(auth_client, name: str = "生成项目") -> dict: @@ -143,3 +143,7 @@ async def scenario(): assert second_queue.empty() asyncio.run(scenario()) + + +def test_generation_response_contract_does_not_expose_user_id(): + assert "user_id" not in GenerationTaskOut.model_fields diff --git a/backend/tests/test_project_api.py b/backend/tests/test_project_api.py index 957ddb75..60ac1efa 100644 --- a/backend/tests/test_project_api.py +++ b/backend/tests/test_project_api.py @@ -30,6 +30,7 @@ def test_create_success(auth_client): assert body["code"] == 200 assert body["message"] == "创建成功" assert body["data"]["id"] is not None + assert "user_id" not in body["data"] assert body["data"]["project_name"] == "新建" assert body["data"]["create_at"] assert "timestamp" not in body @@ -57,7 +58,9 @@ def test_create_validation_error_returns_400(auth_client): def test_get_success(auth_client): - created = auth_client.post("/projects", json=_payload(project_name="详情")).json()["data"] + created = auth_client.post("/projects", json=_payload(project_name="详情")).json()[ + "data" + ] resp = auth_client.get(f"/projects/{created['id']}") assert resp.json()["code"] == 200 @@ -97,14 +100,16 @@ def test_list_paginates(auth_client): assert body["total"] == 3 assert len(body["data"]) == 2 assert [item["project_name"] for item in body["data"]] == ["a2", "a1"] - assert all(item["user_id"] == 1 for item in body["data"]) + assert all("user_id" not in item for item in body["data"]) # -- DELETE /projects/{id} --------------------------------------------------- def test_delete_success(auth_client): - created = auth_client.post("/projects", json=_payload(project_name="删除")).json()["data"] + created = auth_client.post("/projects", json=_payload(project_name="删除")).json()[ + "data" + ] resp = auth_client.delete(f"/projects/{created['id']}") body = resp.json() diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts index a2adc8f7..0f8f9aab 100644 --- a/frontend/src/entities/generation/api.test.ts +++ b/frontend/src/entities/generation/api.test.ts @@ -17,7 +17,6 @@ function success(data: unknown): Response { function taskData(overrides: Record = {}) { return { id: 91, - user_id: 7, project_id: 42, task_type: 'character_image', status: 'completed', @@ -53,7 +52,6 @@ describe('createGenerationApis', () => { const stream = vi.fn(() => vi.fn()) const apis = createGenerationApis({ baseUrl: 'https://api.test/', - userId: '7', transport: { request, stream }, }) @@ -115,7 +113,6 @@ describe('createGenerationApis', () => { ) const apis = createGenerationApis({ baseUrl: '', - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -161,7 +158,6 @@ describe('createGenerationApis', () => { ) const apis = createGenerationApis({ baseUrl: '/api', - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -199,7 +195,6 @@ describe('createGenerationApis', () => { it('拒绝未知任务状态而不是默认为 pending', async () => { const request = vi.fn(async () => success(taskData({ status: 'queued' }))) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -216,7 +211,6 @@ describe('createGenerationApis', () => { success(taskData({ result: { type: 'character_image', image_urls: [null] } })), ) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -236,7 +230,6 @@ describe('createGenerationApis', () => { }) const apis = createGenerationApis({ baseUrl: 'https://api.test', - userId: 7, transport: { request: vi.fn(), stream }, }) const onEvent = vi.fn() @@ -252,7 +245,6 @@ describe('createGenerationApis', () => { const isTerminal = streamOptions?.onEvent( JSON.stringify({ id: 91, - user_id: 7, project_id: 42, task_type: 'character_action', status: 'completed', @@ -305,7 +297,6 @@ describe('createGenerationApis', () => { return vi.fn() }) const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(async () => success(task)), stream }, }) @@ -339,7 +330,6 @@ describe('createGenerationApis', () => { ), ) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -366,7 +356,6 @@ describe('createGenerationApis', () => { ) .mockResolvedValueOnce(success(taskData({ error_message: 'provider failed' }))) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -404,7 +393,6 @@ describe('createGenerationApis', () => { ['非 JSON 响应', new Response('not-json', { status: 502 }), '无法解析的响应'], ])('拒绝%s', async (_label, response, message) => { const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(async () => response), stream: vi.fn(() => vi.fn()), @@ -418,7 +406,6 @@ describe('createGenerationApis', () => { ['任务 id', { id: 0 }, '生成任务 id 无效'], ['任务类型', { task_type: 'video' }, '生成任务 task_type 无效'], ['项目归属', { project_id: 43 }, '生成任务未归属请求中的项目 42'], - ['用户归属', { user_id: 8 }, '生成任务未归属当前用户'], ['请求任务 id', { id: 92 }, '生成任务 ID 与请求的 91 不一致'], ['输入对象', { input_payload: [] }, '生成任务 input_payload 无效'], ['结果对象', { result: [] }, '生成任务 result 无效'], @@ -433,7 +420,6 @@ describe('createGenerationApis', () => { ['完成结果', { result: null }, '完成任务缺少 result'], ])('校验%s', async (_label, overrides, message) => { const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(async () => success(taskData(overrides))), stream: vi.fn(() => vi.fn()), @@ -509,7 +495,6 @@ describe('createGenerationApis', () => { ], ])('拒绝%s', async (_label, result, message) => { const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(async () => success( @@ -544,7 +529,6 @@ describe('createGenerationApis', () => { ) .mockResolvedValueOnce(success(taskData({ status: 'failed', result: null }))) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -564,7 +548,6 @@ describe('createGenerationApis', () => { it('拒绝无缓存阶段的简写订阅并转发显式订阅错误', () => { let onStreamError: ((error: Error) => void) | undefined const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(), stream: vi.fn((_url, options) => { @@ -584,7 +567,6 @@ describe('createGenerationApis', () => { it('校验调用参数、原始响应和任务类型边界', async () => { const apis = createGenerationApis({ - userId: 7, transport: { request: vi .fn() @@ -630,7 +612,6 @@ describe('createGenerationApis', () => { ), ) const apis = createGenerationApis({ - userId: 7, transport: { request, stream: vi.fn(() => vi.fn()) }, }) @@ -649,11 +630,9 @@ describe('createGenerationApis', () => { 'task_update 类型与 character_template 不匹配', ], [JSON.stringify({ ...taskData(), project_id: 43 }), 'task_update 不属于当前项目'], - [JSON.stringify({ ...taskData(), user_id: 8 }), 'task_update 不属于当前用户'], ])('拒绝非法订阅事件', (payload, message) => { let onStreamEvent: ((data: string) => boolean) | undefined const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(), stream: vi.fn((_url, options) => { @@ -671,7 +650,6 @@ describe('createGenerationApis', () => { let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined const onEvent = vi.fn() const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(), stream: vi.fn((_url, options) => { @@ -713,7 +691,6 @@ describe('createGenerationApis', () => { it('拒绝同时存在但不一致的 task_id 与 id', () => { let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(), stream: vi.fn((_url, options) => { @@ -733,7 +710,6 @@ describe('createGenerationApis', () => { let onStreamEvent: ((data: string, eventName?: string) => boolean) | undefined const onEvent = vi.fn() const apis = createGenerationApis({ - userId: 7, transport: { request: vi.fn(), stream: vi.fn((_url, options) => { @@ -764,7 +740,6 @@ describe('createGenerationApis', () => { .mockResolvedValueOnce(success(taskData())) const onEvent = vi.fn() const apis = createGenerationApis({ - userId: 7, pollIntervalMs: 1, transport: { request, diff --git a/frontend/src/entities/generation/api.ts b/frontend/src/entities/generation/api.ts index 5cab6fef..f97f9d59 100644 --- a/frontend/src/entities/generation/api.ts +++ b/frontend/src/entities/generation/api.ts @@ -24,8 +24,6 @@ export interface GenerationTransport { export interface GenerationApiConfig { /** API 前缀；空字符串表示同源。 */ baseUrl?: string - /** 仅用于校验响应归属；请求授权统一由 transport 携带的 token 决定。 */ - userId: string | number transport: GenerationTransport /** SSE 路由不存在时，任务查询兜底的间隔。 */ pollIntervalMs?: number @@ -39,7 +37,6 @@ interface ResponseEnvelope { interface GenerationTaskDto { id: number - userId: number projectId: number taskType: BackendGenerationType status: TaskStatus @@ -152,7 +149,6 @@ function parseTaskDto(value: unknown): GenerationTaskDto { 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), @@ -331,16 +327,12 @@ function inferExpectation(dto: GenerationTaskDto): GenerationExpectation { 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) } @@ -354,13 +346,12 @@ function validateTaskIdentity( function mapTask( value: unknown, expectedProjectId: number, - expectedUserId: number, expectation?: GenerationExpectation, expectedTaskId?: number, ): Generation { const dto = parseTaskDto(value) const resolvedExpectation = expectation ?? inferExpectation(dto) - validateTaskIdentity(dto, expectedProjectId, expectedUserId, resolvedExpectation, expectedTaskId) + validateTaskIdentity(dto, expectedProjectId, resolvedExpectation, expectedTaskId) return { id: String(dto.id), projectId: String(dto.projectId), @@ -425,7 +416,6 @@ function waitForPoll(delayMs: number, signal: AbortSignal): Promise { function mapEvent( value: unknown, expectedProjectId: number, - expectedUserId: number, expectedTaskId: number, expectation: Extract, eventName: string, @@ -444,12 +434,6 @@ function mapEvent( ) { throw new GenerationApiError('task_update 不属于当前项目', 200) } - if ( - value.user_id !== undefined && - dtoPositiveInteger(value.user_id, 'user_id') !== expectedUserId - ) { - throw new GenerationApiError('task_update 不属于当前用户', 200) - } if (value.input_payload !== undefined) { validateInputPayload(dtoNullableRecord(value.input_payload, 'input_payload'), expectation) } @@ -472,11 +456,10 @@ function mapEvent( /** * 创建 Generation 实体适配器。 * - * HTTP/SSE transport 由宿主注入并统一携带 token；`userId` 仅核对响应归属，不会 - * 写入请求或参与后端授权。三个前端阶段在这里收口为后端的两类 GenerationTask。 + * HTTP/SSE transport 由宿主注入并统一携带 token。三个前端阶段在这里收口为 + * 后端的两类 GenerationTask，用户身份不进入适配器契约。 */ export function createGenerationApis(config: GenerationApiConfig): GenerationApis { - const userId = inputPositiveInteger(config.userId, 'userId') const { request, stream } = config.transport const pollIntervalMs = config.pollIntervalMs ?? 1_000 if (!Number.isFinite(pollIntervalMs) || pollIntervalMs < 0) { @@ -495,7 +478,7 @@ export function createGenerationApis(config: GenerationApiConfig): GenerationApi headers: { 'content-type': 'application/json' }, body: JSON.stringify(body), }) - return mapTask(await readData(response), projectId, userId, expectation) as Generation + return mapTask(await readData(response), projectId, expectation) as Generation } const apis: GenerationApis = { @@ -555,7 +538,7 @@ export function createGenerationApis(config: GenerationApiConfig): GenerationApi ) const raw = await readData(response) const resolvedExpectation = expectation ?? inferExpectation(parseTaskDto(raw)) - const generation = mapTask(raw, numericProjectId, userId, resolvedExpectation, numericTaskId) + const generation = mapTask(raw, numericProjectId, resolvedExpectation, numericTaskId) expectations.set(generation.id, resolvedExpectation) return generation }, @@ -622,7 +605,6 @@ export function createGenerationApis(config: GenerationApiConfig): GenerationApi const event = mapEvent( parseEventData(data), numericProjectId, - userId, numericTaskId, expectation, eventName, diff --git a/frontend/src/entities/project/index.test.ts b/frontend/src/entities/project/index.test.ts index 559b563a..d979bb10 100644 --- a/frontend/src/entities/project/index.test.ts +++ b/frontend/src/entities/project/index.test.ts @@ -2,7 +2,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest' const projectDto = { id: 42, - user_id: 7, workflow_id: 9, project_name: '点灯人', character_perspective: 3, @@ -51,11 +50,10 @@ describe('projectApis', () => { ) }) - await expect(projectApis.list({ page: 2, pageSize: 10, ownerId: '7' })).resolves.toEqual({ + await expect(projectApis.list({ page: 2, pageSize: 10 })).resolves.toEqual({ items: [ { id: '42', - ownerId: '7', workflowId: '9', name: '点灯人', perspective: 'isometric', @@ -71,7 +69,7 @@ describe('projectApis', () => { page: 2, pageSize: 10, }) - expect(request?.url).toBe('https://api.windup.test/projects?page=2&page_size=10&user_id=7') + expect(request?.url).toBe('https://api.windup.test/projects?page=2&page_size=10') }) it('serializes CreateProjectInput to the backend request body', async () => { diff --git a/frontend/src/entities/project/index.ts b/frontend/src/entities/project/index.ts index 7ad4ecf2..23f8af60 100644 --- a/frontend/src/entities/project/index.ts +++ b/frontend/src/entities/project/index.ts @@ -4,7 +4,6 @@ import type { Paged, PageQuery } from '@/shared/pagination' /** Project 前端领域形状；字段名由本模块显式映射后端 ProjectOut。 */ export interface Project { id: string - ownerId: string workflowId: string | null name: string perspective: CharacterPerspective @@ -18,8 +17,7 @@ export interface Project { /** * 后端创建 Project 需要的完整字段。 - * 这里没有 ownerId：归属由后端从 access token 里取（`ProjectCreate` 不含 user_id）， - * 请求体再带一个用户 ID 就等于宣称调用方可以替别人建项目。 + * 项目归属由后端从 access token 读取，不进入前端请求契约。 */ export interface CreateProjectInput { workflowId?: string | null @@ -31,10 +29,7 @@ export interface CreateProjectInput { sampleImageUrl?: string | null } -export interface ProjectPageQuery extends PageQuery { - /** 对应后端 user_id；后端按登录用户强制隔离前保持可选。 */ - ownerId?: string -} +export type ProjectPageQuery = PageQuery /** 后端 character_perspective: 1 横版 / 2 俯视 / 3 2.5D。 */ export type CharacterPerspective = 'side' | 'top-down' | 'isometric' @@ -64,7 +59,6 @@ export interface ProjectApis { interface ProjectDto { id: number - user_id: number workflow_id: number | null project_name: string character_perspective: number @@ -119,7 +113,6 @@ function toBackendId(value: string, field: string): number { function mapProject(dto: ProjectDto): Project { return { id: String(dto.id), - ownerId: String(dto.user_id), workflowId: dto.workflow_id === null ? null : String(dto.workflow_id), name: dto.project_name, perspective: mapEnumValue( @@ -150,7 +143,6 @@ export const projectApis: ProjectApis = { query: { page: query.page, page_size: query.pageSize, - user_id: query.ownerId ? toBackendId(query.ownerId, 'ownerId') : undefined, }, }) return { ...result, items: result.items.map(mapProject) } diff --git a/frontend/src/pages/workflow-editor/index.test.tsx b/frontend/src/pages/workflow-editor/index.test.tsx index 7b9b9e96..4a526fcb 100644 --- a/frontend/src/pages/workflow-editor/index.test.tsx +++ b/frontend/src/pages/workflow-editor/index.test.tsx @@ -1001,7 +1001,6 @@ function deferred(): Deferred { function projectFixture(): Project { return { id: '1', - ownerId: '7', workflowId: null, name: '正式项目', perspective: 'side', diff --git a/frontend/src/pages/workflow-editor/runtime.test.ts b/frontend/src/pages/workflow-editor/runtime.test.ts index c3a009d9..7dff6fa4 100644 --- a/frontend/src/pages/workflow-editor/runtime.test.ts +++ b/frontend/src/pages/workflow-editor/runtime.test.ts @@ -235,7 +235,6 @@ function workflowFixture(): WorkflowRun { function projectFixture(): Project { return { id: '1', - ownerId: '7', workflowId: null, name: '正式项目', perspective: 'side', diff --git a/frontend/src/test/project-assets-backend.ts b/frontend/src/test/project-assets-backend.ts index 0d65afad..d60bd54c 100644 --- a/frontend/src/test/project-assets-backend.ts +++ b/frontend/src/test/project-assets-backend.ts @@ -1,7 +1,6 @@ /** 后端 ProjectOut 的形状；写死成显式类型，免得 fixture 的字面量把可空字段收窄。 */ interface ProjectDto { id: number - user_id: number workflow_id: number | null project_name: string character_perspective: number @@ -17,7 +16,6 @@ interface ProjectDto { const projectDtos: ProjectDto[] = [ { id: 42, - user_id: 7, workflow_id: null, project_name: '点灯人 · MVP', character_perspective: 1, @@ -31,7 +29,6 @@ const projectDtos: ProjectDto[] = [ }, { id: 99, - user_id: 7, workflow_id: null, project_name: '空白海岸', character_perspective: 2, @@ -224,8 +221,6 @@ export function createProjectAssetsBackend({ } const created = { id: 4_242, - // 后端从 access token 取归属，请求体里没有 user_id；这里跟 fixture 用同一个用户。 - user_id: 7, workflow_id: body.workflow_id ?? null, project_name: body.project_name, character_perspective: body.character_perspective,