From 0d965451d8bd51444caa6136e4be1768021250fb Mon Sep 17 00:00:00 2001 From: xyh202131 <246811510+xyh202131@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:05:26 +0800 Subject: [PATCH] feat(generation): add SSE task adapter --- "_PR\350\257\264\346\230\216.md" | 70 +++ frontend/src/entities/generation/api.test.ts | 349 +++++++++++++ frontend/src/entities/generation/api.ts | 506 +++++++++++++++++++ frontend/src/entities/generation/index.ts | 50 +- frontend/src/entities/index.ts | 5 + frontend/src/shared/api/stream.test.ts | 108 ++++ frontend/src/shared/api/stream.ts | 89 ++++ 7 files changed, 1168 insertions(+), 9 deletions(-) create mode 100644 "_PR\350\257\264\346\230\216.md" 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/_PR\350\257\264\346\230\216.md" "b/_PR\350\257\264\346\230\216.md" new file mode 100644 index 0000000..14a1c59 --- /dev/null +++ "b/_PR\350\257\264\346\230\216.md" @@ -0,0 +1,70 @@ +# Generation + SSE Adapter + +Refs #78 + +Issue #78 已补 `mile3` 标签并关联 `milestone/4`。本变更只实现前端 Generation +实体适配器与 SSE 传输封装，不修改后端、页面、Controller、共享入口或构建产物。 + +## 变更范围 + +- 将创建、按项目查询和状态订阅统一收口到 `GenerationApis`。 +- `createGenerationApis` 必须由宿主注入 `userId` 与 `transport`；模块不写死用户身份， + 也不直接持有 `fetch` 或 `EventSource`。 +- 新增业务无关的 `shared/api/stream.ts`，封装命名 SSE 事件、取消、终态关闭、 + 非法消息错误和断线通知。临时断线保留浏览器 EventSource 的协议级自动重连， + 不恢复 2 秒业务轮询。 +- 查询与订阅要求调用方传入 WorkflowRun 已知的阶段期望；动作阶段还必须带上 + `actionType`，避免把其他动作的帧误接到当前任务。 +- 角色母版由宿主通过 `resolveImageSize(projectId)` 提供项目画布尺寸；Generation 不直接 + 依赖 Project，也不会回退到可能冲突的 1024 默认值。 + +## 三阶段合同 + +| 前端阶段 | 后端请求 | 固定/可配置数量 | 结果映射 | +| -------------------- | ------------------------- | ------------------------- | ---------------------------- | +| `character_template` | `POST /generation/image` | 固定 `num_images: 4` | 严格校验并映射 4 个候选 | +| `first_frame` | `POST /generation/action` | 固定 `num_frames: 1` | 严格校验并映射 1 帧动作首帧 | +| `complete_animation` | `POST /generation/action` | 当前固定 `num_frames: 16` | 按后端 `frames[].index` 排序 | + +`CompleteAnimationGenerationInput` 当前没有 `frameCount` 字段，因此无法由调用输入表达 +帧数。本次保守沿用后端合同默认值 16；后续若合同加入可配置字段，应改为透传输入并补充 +边界校验，而不是继续保留常量。 + +## SSE 行为 + +- 端点：`/generation/tasks/{taskId}/stream?project_id=...`。 +- 只监听 `task_update` 命名事件，事件 DTO 在 Generation 边界解析和校验。 +- `completed`、`failed` 都视为终态；事件先交付调用方，再由传输层关闭连接。 +- `onError` 必传，非法事件关闭连接时不能静默留下永远等待的工作流。 +- 显式取消返回幂等函数，并移除监听器、清空错误处理器、关闭 EventSource。 +- 非法 JSON、非法 DTO 或调用方事件处理异常会报告错误并关闭连接。 +- 浏览器连接中断会报告 `SSE 连接中断`，连接保持给 EventSource 自动重连；没有定时 + GET、退避 GET 或其他业务轮询。 + +## DTO 校验 + +- 校验响应 envelope 的 `code/message/data`，业务错误不会被当作成功数据。 +- 校验任务与事件的正整数 ID、项目归属、用户归属、任务类型和四个合法状态。 +- 未知状态直接抛 `GenerationApiError`，绝不降级为 `pending`。 +- 校验完成结果的判别字段、图片 URL、候选/首帧数量、动作类型、16 帧数量与连续索引、 + 时长格式与状态/错误一致性；非完成任务携带结果同样视为非法合同。 + +## 测试覆盖 + +- 三阶段请求体映射与注入用户身份。 +- 角色母版四候选、动作一首帧、完整动画帧排序。 +- 未知状态与非法完成结果 DTO。 +- SSE URL、`task_update` 映射、主动取消、终态关闭、事件解析错误和断线恢复语义。 + +## 验证结果 + +- `npx oxfmt --check` 定向检查本次 5 个 TypeScript 文件：通过。 +- `npm run format:check` 全量检查：已执行，但被基线中 41 个本次所有权外文件阻断； + 未批量重写这些文件，以免覆盖其他工作者的修改。 +- `npm run lint`：通过。 +- `npm run typecheck`：通过。 +- `npm test`：通过，4 个测试文件、12 个测试，其中本模块新增 10 个。 +- `npm run build`：通过，Vite 8.1.5 共转换 88 个模块。 + +后端 `/generation/tasks/{taskId}/stream` 的实现与真实联调不在本前端-only 变更范围内； +本次测试通过注入 transport 验证前端合同，不把它表述为后端 SSE 已可用。 diff --git a/frontend/src/entities/generation/api.test.ts b/frontend/src/entities/generation/api.test.ts new file mode 100644 index 0000000..434612e --- /dev/null +++ b/frontend/src/entities/generation/api.test.ts @@ -0,0 +1,349 @@ +import { describe, expect, it, vi } from 'vitest' + +import { createGenerationApis, GenerationApiError } from '@/entities' + +import type { MediaReference } from '../media' + +const reference = (url: string) => url as MediaReference +const resolveImageSize = vi.fn(async () => ({ width: 64, height: 96 })) + +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 }, + resolveImageSize, + }) + + const generation = await apis.create({ + type: 'character_template', + projectId: '42', + referenceMedia: [reference('https://cdn.test/reference.png')], + prompt: 'pixel hero', + }) + + expect(request).toHaveBeenCalledWith( + 'https://api.test/generation/image', + expect.objectContaining({ + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + user_id: 7, + 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' }, + ], + }) + expect(resolveImageSize).toHaveBeenCalledWith('42') + }) + + 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()) }, + resolveImageSize, + }) + + 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({ + user_id: 7, + 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: 16, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(16), + }, + }), + ), + ) + const apis = createGenerationApis({ + baseUrl: '/api', + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + resolveImageSize, + }) + + 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({ + user_id: 7, + 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: 16, + }) + expect(generation.result).toEqual({ + type: 'complete_animation', + frames: Array.from({ length: 16 }, (_, index) => ({ + url: `https://cdn.test/frame-${index + 1}.png`, + })), + }) + }) + + it('拒绝未知任务状态而不是默认为 pending', async () => { + const request = vi.fn(async () => success(taskData({ status: 'queued' }))) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + resolveImageSize, + }) + + 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()) }, + resolveImageSize, + }) + + 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 }, + resolveImageSize, + }) + 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({ + task_id: 91, + task_type: 'character_action', + status: 'completed', + result: { + type: 'character_action', + action_type: 'walk', + frames: actionFrames(16), + }, + 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: 16 }, (_, index) => ({ + url: `https://cdn.test/frame-${index + 1}.png`, + })), + }, + error: null, + }) + + unsubscribe() + expect(cancel).toHaveBeenCalledOnce() + }) + + it('拒绝 completed 任务返回错误动作类型', async () => { + const request = vi.fn(async () => + success( + taskData({ + task_type: 'character_action', + input_payload: { num_frames: 16, action_type: 'walk' }, + result: { + type: 'character_action', + action_type: 'attack', + frames: actionFrames(16), + }, + }), + ), + ) + const apis = createGenerationApis({ + userId: 7, + transport: { request, stream: vi.fn(() => vi.fn()) }, + resolveImageSize, + }) + + 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: 16, 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()) }, + resolveImageSize, + }) + + await expect( + apis.get('42', '91', { type: 'complete_animation', actionType: 'walk' }), + ).rejects.toThrow('完整动画结果必须包含 16 帧') + 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 0000000..68b68ac --- /dev/null +++ b/frontend/src/entities/generation/api.ts @@ -0,0 +1,506 @@ +import type { EventStreamSubscriber } from '@/shared/api/stream' + +import type { + CompleteAnimationGenerationInput, + GeneratedImage, + Generation, + GenerationApis, + GenerationEvent, + GenerationExpectation, + GenerationImageSize, + 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 + /** 由组合根通过 ProjectApis 提供，Generation 不直接依赖 Project 实体或猜测尺寸。 */ + resolveImageSize(projectId: string): Promise +} + +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, + image: { url: nonEmptyString(frame.image_url, '动作帧 image_url') }, + } + }) + + const orderedFrames = frames + .sort((left, right) => left.index - right.index) + .map(({ image }) => image) + const expectedFrameCount = expectation.type === 'first_frame' ? 1 : 16 + 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: orderedFrames[0]! } + } + return { type: 'complete_animation', frames: orderedFrames } +} + +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 : 16 + if (inputPayload.num_frames !== expectedFrameCount) { + throw new GenerationApiError( + `动作任务 input_payload.num_frames 必须为 ${expectedFrameCount}`, + 200, + ) + } + if (inputPayload.action_type !== expectation.actionType) { + throw new GenerationApiError('动作任务 input_payload.action_type 与请求不一致', 200) + } +} + +function validateTaskIdentity( + dto: GenerationTaskDto, + expectedProjectId: number, + expectedUserId: number, + expectation: GenerationExpectation, + expectedTaskId?: number, +): void { + if (dto.projectId !== expectedProjectId) { + throw new GenerationApiError(`生成任务未归属请求中的项目 ${expectedProjectId}`, 200) + } + if (dto.userId !== expectedUserId) { + throw new GenerationApiError('生成任务未归属当前用户', 200) + } + if (expectedTaskId !== undefined && dto.id !== expectedTaskId) { + throw new GenerationApiError(`生成任务 ID 与请求的 ${expectedTaskId} 不一致`, 200) + } + if (dto.taskType !== expectedBackendType(expectation.type)) { + throw new GenerationApiError(`生成任务类型与 ${expectation.type} 不匹配`, 200) + } + validateStatusError(dto.status, dto.errorMessage) + validateInputPayload(dto.inputPayload, expectation) +} + +function mapTask( + value: unknown, + expectedProjectId: number, + expectedUserId: number, + expectation: Extract, + expectedTaskId?: number, +): Generation { + const dto = parseTaskDto(value) + validateTaskIdentity(dto, expectedProjectId, expectedUserId, expectation, expectedTaskId) + return { + id: String(dto.id), + projectId: String(dto.projectId), + type: expectation.type, + status: dto.status, + result: mapResult(dto.result, dto.status, expectation), + error: dto.errorMessage, + } +} + +function references(input: CompleteAnimationGenerationInput): string[] { + return [input.firstFrameUrl, ...input.referenceMedia.map(String)].filter( + (url, index, all) => url.trim() !== '' && all.indexOf(url) === index, + ) +} + +function parseEventData(data: string): unknown { + try { + return JSON.parse(data) as unknown + } catch (error) { + throw new GenerationApiError('task_update 不是有效 JSON', 200, { cause: error }) + } +} + +function mapEvent( + value: unknown, + expectedTaskId: number, + expectation: Extract, +): GenerationEvent { + if (!isRecord(value)) throw new GenerationApiError('task_update 不是对象', 200) + const taskId = dtoPositiveInteger(value.task_id, 'task_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) + } + const status = taskStatus(value.status) + const result = dtoNullableRecord(value.result ?? null, 'result') + const error = dtoNullableString(value.error_message ?? null, '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/EventSource。三个前端阶段在这里收口为后端的两类 GenerationTask。 + */ +export function createGenerationApis(config: GenerationApiConfig): GenerationApis { + const userId = inputPositiveInteger(config.userId, 'userId') + const { request, stream } = config.transport + + 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) + } + + return { + async create(input: T): Promise> { + const projectId = inputPositiveInteger(input.projectId, 'projectId') + if (input.type !== 'character_template') { + const referenceImageUrls = + input.type === 'complete_animation' + ? references(input) + : input.referenceMedia.map(String).filter((url) => url.trim() !== '') + return post( + '/generation/action', + projectId, + { type: input.type, actionType: input.actionType }, + { + user_id: userId, + 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, + // 首帧是一次一帧动作任务；完整动画当前沿用后端合同的 16 帧。 + num_frames: input.type === 'first_frame' ? 1 : 16, + }, + ) + } + + const imageSize = await config.resolveImageSize(input.projectId) + return post( + '/generation/image', + projectId, + { type: input.type }, + { + user_id: userId, + project_id: projectId, + reference_image_url: input.referenceMedia[0] ? String(input.referenceMedia[0]) : null, + prompt: input.prompt ?? '', + negative_prompt: '', + width: inputPositiveInteger(imageSize.width, 'imageSize.width'), + height: inputPositiveInteger(imageSize.height, 'imageSize.height'), + // 只有角色母版走图片接口，并且固定生成四个候选。 + num_images: 4, + }, + ) + }, + + async get( + projectId: string, + id: string, + expectation: Extract, + ): Promise> { + const numericProjectId = inputPositiveInteger(projectId, 'projectId') + const numericTaskId = inputPositiveInteger(id, 'taskId') + const response = await request( + endpoint( + config.baseUrl, + `/generation/tasks/${numericTaskId}?project_id=${numericProjectId}`, + ), + { method: 'GET' }, + ) + return mapTask(await readData(response), numericProjectId, userId, expectation, numericTaskId) + }, + + subscribe( + projectId: string, + id: string, + expectation: Extract, + onEvent: (event: GenerationEvent) => void, + onError: (error: Error) => void, + ): () => void { + const numericProjectId = inputPositiveInteger(projectId, 'projectId') + const numericTaskId = inputPositiveInteger(id, 'taskId') + return stream( + endpoint( + config.baseUrl, + `/generation/tasks/${numericTaskId}/stream?project_id=${numericProjectId}`, + ), + { + eventName: 'task_update', + onEvent(data) { + const event = mapEvent(parseEventData(data), numericTaskId, expectation) + onEvent(event) + return event.status === 'completed' || event.status === 'failed' + }, + onError, + }, + ) + }, + } +} diff --git a/frontend/src/entities/generation/index.ts b/frontend/src/entities/generation/index.ts index cadb63d..b5b0524 100644 --- a/frontend/src/entities/generation/index.ts +++ b/frontend/src/entities/generation/index.ts @@ -19,25 +19,41 @@ export type TaskStatus = 'pending' | 'running' | 'completed' | 'failed' /** * 生成对应的三个前端可见异步步骤。 * 它是前端工作流粒度，不等于后端 task_type——后端只有 character_image 与 - * character_action 两种，character_template 和 first_frame 都落在 character_image 上。 + * character_action 两种：character_template 落在 character_image，动作首帧和完整动画 + * 都落在 character_action，只是请求帧数分别为 1 和 16。 * 完整动画内部可含视频生成、截帧和多次图像处理，但对前端仍是一次 Generation。 */ export type GenerationType = 'character_template' | 'first_frame' | 'complete_animation' +/** + * 恢复任务时由 WorkflowRun 提供的已知上下文。动作阶段必须带上动作语义， + * 这样适配器才能拒绝“请求 walk、后端却返回 attack”这类串任务结果。 + */ +export type GenerationExpectation = + | { type: 'character_template' } + | { type: 'first_frame'; actionType: ActionType } + | { type: 'complete_animation'; actionType: ActionType } + interface GenerationInputBase { projectId: string /** 可选参考媒体；没有参考图时传空数组。 */ referenceMedia: readonly MediaReference[] } -/** 角色母版候选生成。 */ +/** 图片生成请求的实际画布尺寸，必须与所属项目的精灵尺寸合同一致。 */ +export interface GenerationImageSize { + width: number + height: number +} + +/** 角色母版候选生成；当前合同固定请求 4 个候选，不向调用方暴露可变数量。 */ export interface CharacterTemplateGenerationInput extends GenerationInputBase { type: 'character_template' /** 已由手动输入或 Quick Start 整理好的角色提示词。 */ prompt: string } -/** 指定角色造型下的动作首帧生成；不能只绑定 Character。 */ +/** 指定角色造型下的动作首帧生成；当前合同固定 1 张，且不能只绑定 Character。 */ export interface FirstFrameGenerationInput extends GenerationInputBase { type: 'first_frame' characterId: string @@ -47,7 +63,10 @@ export interface FirstFrameGenerationInput extends GenerationInputBase { prompt: string | null } -/** 以已确认首帧为起点生成完整动画。 */ +/** + * 以已确认首帧为起点生成完整动画。 + * 后端支持 num_frames，但当前输入没有 frameCount；适配器暂按后端默认值提交 16。 + */ export interface CompleteAnimationGenerationInput extends GenerationInputBase { type: 'complete_animation' characterId: string @@ -101,7 +120,8 @@ export type GenerationResultFor = * 一次生成任务的完整快照，创建、查询和断线恢复都用它。 * 它是服务端的资源，不是一次「调用能力」——前端创建它，然后订阅或轮询它的状态。 * - * TType 在调用边界已知时保留精确类型；按 ID 恢复时用默认值，等运行时解析后再收窄。 + * TType 在调用边界已知时保留精确类型；查询和订阅必须传入工作流已知的前端阶段， + * 因为后端 task_type 比前端阶段更粗，不能只靠 DTO 猜测首帧还是完整动画。 * 完成不代表工作流节点已通过，节点状态由 WorkflowStep 自己判定。 */ export interface Generation { @@ -137,11 +157,23 @@ export interface GenerationApis { * 按所属项目和任务 ID 读取最新快照。 * projectId 不能从 id 推导，后端查询接口要求两者同时传入。 */ - get(projectId: Generation['projectId'], id: Generation['id']): Promise - /** 订阅状态变化，返回取消订阅函数。 */ - subscribe( + get( + projectId: Generation['projectId'], + id: Generation['id'], + expectation: Extract, + ): Promise> + /** + * 订阅 task_update，终态由传输层自动关闭；返回的函数供页面离开时主动取消。 + * 传输错误与非法 DTO 通过 onError 上报，不伪造成业务 failed 状态。 + */ + subscribe( projectId: Generation['projectId'], id: Generation['id'], - onEvent: (event: GenerationEvent) => void, + 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 203359d..f58f2b7 100644 --- a/frontend/src/entities/index.ts +++ b/frontend/src/entities/index.ts @@ -34,6 +34,7 @@ export type { export type { ActionTemplate, ActionTemplateApis } from './action-template' /* 生成 —— 业务数据，不是「调用生成能力」；后端的 task 就是它，不另立实体 */ +export { createGenerationApis, GenerationApiError } from './generation' export type { CharacterTemplateGenerationInput, CharacterTemplateGenerationResult, @@ -45,10 +46,14 @@ export type { Generation, GenerationApis, GenerationEvent, + GenerationExpectation, GenerationInput, + GenerationImageSize, GenerationResult, GenerationResultFor, GenerationType, + GenerationApiConfig, + GenerationTransport, TaskStatus, } from './generation' diff --git a/frontend/src/shared/api/stream.test.ts b/frontend/src/shared/api/stream.test.ts new file mode 100644 index 0000000..2590849 --- /dev/null +++ b/frontend/src/shared/api/stream.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it, vi } from 'vitest' + +import { subscribeToEventStream, type EventSourceFactory, type EventSourceLike } from './stream' + +class FakeEventSource implements EventSourceLike { + readonly listeners = new Map void>>() + readonly close = vi.fn() + onerror: ((event: Event) => void) | null = null + + addEventListener(type: string, listener: (event: Event) => void): void { + const listeners = this.listeners.get(type) ?? new Set() + listeners.add(listener) + this.listeners.set(type, listeners) + } + + removeEventListener(type: string, listener: (event: Event) => void): void { + this.listeners.get(type)?.delete(listener) + } + + emit(type: string, data: string): void { + for (const listener of this.listeners.get(type) ?? []) { + listener({ data } as MessageEvent) + } + } + + disconnect(): void { + this.onerror?.(new Event('error')) + } +} + +function setup() { + const source = new FakeEventSource() + const factory = vi.fn(() => source) + return { source, factory } +} + +describe('subscribeToEventStream', () => { + it('取消后关闭连接并忽略后续事件', () => { + const { source, factory } = setup() + const onEvent = vi.fn(() => false) + const unsubscribe = subscribeToEventStream( + '/generation/tasks/91/stream?project_id=42', + { eventName: 'task_update', onEvent, onError: vi.fn() }, + factory, + ) + + unsubscribe() + source.emit('task_update', '{"status":"running"}') + + expect(source.close).toHaveBeenCalledOnce() + expect(onEvent).not.toHaveBeenCalled() + }) + + it('收到终态关闭信号后关闭连接且只交付一次', () => { + const { source, factory } = setup() + const onEvent = vi.fn(() => true) + subscribeToEventStream( + '/generation/tasks/91/stream?project_id=42', + { eventName: 'task_update', onEvent, onError: vi.fn() }, + factory, + ) + + source.emit('task_update', '{"status":"completed"}') + source.emit('task_update', '{"status":"completed"}') + + expect(onEvent).toHaveBeenCalledOnce() + expect(source.close).toHaveBeenCalledOnce() + }) + + it('事件解析失败时报告错误并关闭连接', () => { + const { source, factory } = setup() + const onError = vi.fn() + subscribeToEventStream( + '/generation/tasks/91/stream?project_id=42', + { + eventName: 'task_update', + onEvent: () => { + throw new Error('invalid task DTO') + }, + onError, + }, + factory, + ) + + source.emit('task_update', '{}') + + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message: 'invalid task DTO' })) + expect(source.close).toHaveBeenCalledOnce() + }) + + it('断线时报告错误并保留 EventSource 的自动重连能力', () => { + const { source, factory } = setup() + const onEvent = vi.fn(() => false) + const onError = vi.fn() + subscribeToEventStream( + '/generation/tasks/91/stream?project_id=42', + { eventName: 'task_update', onEvent, onError }, + factory, + ) + + source.disconnect() + source.emit('task_update', '{"status":"running"}') + + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message: 'SSE 连接中断' })) + expect(source.close).not.toHaveBeenCalled() + expect(onEvent).toHaveBeenCalledOnce() + }) +}) diff --git a/frontend/src/shared/api/stream.ts b/frontend/src/shared/api/stream.ts new file mode 100644 index 0000000..9308102 --- /dev/null +++ b/frontend/src/shared/api/stream.ts @@ -0,0 +1,89 @@ +/** + * 业务无关的 SSE 订阅边界。 + * + * 上层只处理字符串 payload、终态判断和错误回调，不接触 EventSource 实例。 + * 浏览器断线后由 EventSource 按协议自动重连；显式取消、终态或非法消息才会关闭连接。 + */ + +export interface EventSourceLike { + onerror: ((event: Event) => void) | null + addEventListener(type: string, listener: (event: Event) => void): void + removeEventListener(type: string, listener: (event: Event) => void): void + close(): void +} + +export type EventSourceFactory = (url: string) => EventSourceLike + +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 class EventStreamError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options) + this.name = 'EventStreamError' + } +} + +const createBrowserEventSource: EventSourceFactory = (url) => new EventSource(url) + +function asError(value: unknown): Error { + return value instanceof Error ? value : new EventStreamError('SSE 事件处理失败') +} + +/** + * 建立命名 SSE 事件订阅并返回幂等取消函数。 + * + * `error` 事件通常表示临时断线。此处只通知上层而不主动 close，让浏览器原生 + * EventSource 继续使用服务端 retry 配置重连，避免退回业务轮询。 + */ +export function subscribeToEventStream( + url: string, + options: EventStreamOptions, + eventSourceFactory: EventSourceFactory = createBrowserEventSource, +): () => void { + let active = true + let source: EventSourceLike | null = null + + const stop = () => { + if (!active) return + active = false + if (source === null) return + source.removeEventListener(options.eventName, handleEvent) + source.onerror = null + source.close() + } + + const handleEvent = (event: Event) => { + if (!active) return + try { + if (!('data' in event) || typeof event.data !== 'string') { + throw new EventStreamError('SSE 事件缺少字符串 data') + } + if (options.onEvent(event.data)) stop() + } catch (error) { + stop() + options.onError(asError(error)) + } + } + + try { + source = eventSourceFactory(url) + source.addEventListener(options.eventName, handleEvent) + source.onerror = () => { + if (active) options.onError(new EventStreamError('SSE 连接中断')) + } + } catch (error) { + stop() + options.onError(new EventStreamError('SSE 连接建立失败', { cause: error })) + } + + return stop +}