diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index 5e8ad75f6..364977f72 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -172,6 +172,26 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, task) }) + r.POST("/tasks/:id/text-replay-complete", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + var req struct { + Text string `json:"text"` + } + if err := c.ShouldBindJSON(&req); err != nil { + fail(c, http.StatusBadRequest, err) + return + } + task, err := svc.CompleteTextReplayTask(user.ID, c.Param("id"), req.Text) + if err != nil { + fail(c, http.StatusBadRequest, err) + return + } + ok(c, task) + }) r.GET("/tasks/:id/logs", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 4b4f72008..93d5fb37d 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -44,6 +44,10 @@ const ( TaskStatusSucceeded TaskStatus = "succeeded" TaskStatusFailed TaskStatus = "failed" TaskStatusCancelled TaskStatus = "cancelled" + // text_replay 任务是前端自管、不经过 worker 队列的文本生成存档容器: + // 前端直连模型流式生成,把增量 POST 到 /tasks/:id/text-deltas 存档, + // 完成后调用 /tasks/:id/text-replay-complete 归并为最终正文。 + TaskStatusTextReplay TaskStatus = "text_replay" SessionStatusActive SessionStatus = "active" SessionStatusCompleted SessionStatus = "completed" diff --git a/backend/internal/repository/text_replay.go b/backend/internal/repository/text_replay.go index 26b51c7b0..90aa833d9 100644 --- a/backend/internal/repository/text_replay.go +++ b/backend/internal/repository/text_replay.go @@ -124,6 +124,20 @@ func (r *Repository) CleanupTaskTextDeltas(now time.Time) (int64, error) { return deleted, err } +// CompleteTextReplayTask 把前端自管的 text_replay 任务原子收尾:写入最终正文并置为 succeeded。 +// 条件更新保证只有仍处于 text_replay 状态的任务能被收尾,避免重复完成。 +func (r *Repository) CompleteTextReplayTask(userID string, taskID string, resultJSON string, now time.Time) (bool, error) { + result := r.db.Model(&model.Task{}). + Where("id = ? AND user_id = ? AND status = ?", taskID, userID, model.TaskStatusTextReplay). + Updates(map[string]any{ + "status": model.TaskStatusSucceeded, "stage": "已完成", "progress": 100, + "result_json": resultJSON, "text_draft": "", + "error": "", "completed_at": &now, + "lease_owner": "", "lease_expires_at": nil, "updated_at": now, + }) + return result.RowsAffected == 1, result.Error +} + func (r *Repository) DeleteUserTaskTextDeltas(userID string) error { return r.db.Delete(&model.TaskTextDelta{}, "user_id = ?", userID).Error } diff --git a/backend/internal/repository/text_replay_test.go b/backend/internal/repository/text_replay_test.go new file mode 100644 index 000000000..f09f4878d --- /dev/null +++ b/backend/internal/repository/text_replay_test.go @@ -0,0 +1,91 @@ +package repository + +import ( + "testing" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +// 验证 text-replay 闭环:text_replay 任务可写增量,完成后置为 succeeded 并写入最终正文。 +func TestTextReplayLifecycle(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:text-replay-lifecycle?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Task{}, &model.TaskTextDelta{}); err != nil { + t.Fatal(err) + } + task := model.Task{ + ID: "replay-1", UserID: "user-1", Type: "text", Prompt: "写一段剧本", + Status: model.TaskStatusTextReplay, Stage: "文本持久化(前端自管)", + } + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + + repo := New(db) + now := time.Now() + limits := TextReplayLimits{MaxTaskBytes: 1 << 20, MaxUserBytes: 1 << 20, MaxTaskEvents: 100} + + // text_replay 状态应允许写入增量(未结束) + if _, err := repo.AppendTaskTextDelta("user-1", "replay-1", "第一段内容", now.Add(time.Hour), limits); err != nil { + t.Fatalf("text_replay 任务写入增量失败: %v", err) + } + if _, err := repo.AppendTaskTextDelta("user-1", "replay-1", "第二段内容", now.Add(time.Hour), limits); err != nil { + t.Fatalf("text_replay 任务写入增量失败: %v", err) + } + + // 完成:写入最终正文并置为 succeeded + resultJSON := `{"mode":"text","text":"第一段内容第二段内容"}` + completed, err := repo.CompleteTextReplayTask("user-1", "replay-1", resultJSON, now) + if err != nil { + t.Fatal(err) + } + if !completed { + t.Fatal("expected complete to succeed") + } + stored, err := repo.Task("replay-1") + if err != nil { + t.Fatal(err) + } + if stored.Status != model.TaskStatusSucceeded { + t.Fatalf("expected status succeeded, got %s", stored.Status) + } + if stored.ResultJSON != resultJSON { + t.Fatalf("unexpected result_json: %s", stored.ResultJSON) + } + if stored.CompletedAt == nil { + t.Fatal("expected completed_at set") + } + + // 已完成的 text_replay 任务不能再写入增量(进入 closed) + if _, err := repo.AppendTaskTextDelta("user-1", "replay-1", "追加", now.Add(time.Hour), limits); err == nil { + t.Fatal("expected ErrTextReplayClosed after completion") + } + + // 重复完成应返回 completed=false(状态已不是 text_replay) + again, err := repo.CompleteTextReplayTask("user-1", "replay-1", resultJSON, now.Add(time.Minute)) + if err != nil { + t.Fatal(err) + } + if again { + t.Fatal("expected second complete to be a no-op") + } + + // 非 text_replay 任务不能完成 + other := model.Task{ID: "queued-1", UserID: "user-1", Type: "text", Status: model.TaskStatusQueued} + if err := db.Create(&other).Error; err != nil { + t.Fatal(err) + } + done, err := repo.CompleteTextReplayTask("user-1", "queued-1", resultJSON, now) + if err != nil { + t.Fatal(err) + } + if done { + t.Fatal("queued task must not be completable as text_replay") + } +} diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index b14b20c43..b9097073c 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -347,6 +347,10 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task, } normalizedInput = applyRoutedProviderSelection(normalizedInput, routed) } + // 前端自管的文本持久化任务:直连模型生成、增量上报 text-deltas,不排入 worker 队列生成。 + if isTextReplayTaskRequest(normalizedInput) { + return s.createTextReplayTask(userID, req, normalizedInput) + } if err := s.requireCustomChannelsForTaskInput(normalizedInput); err != nil { return nil, err } @@ -473,6 +477,41 @@ func normalizeTaskInput(input map[string]any) (map[string]any, error) { return normalized, nil } +// createTextReplayTask 创建前端自管的文本持久化任务:状态为 text_replay, +// 不排队执行、不计 active 队列、不产生计费,仅作为正文增量(text-deltas)的存储容器。 +func (s *Service) createTextReplayTask(userID string, req CreateTaskRequest, normalizedInput map[string]any) (*model.Task, error) { + prompt := strings.TrimSpace(req.Prompt) + if prompt == "" { + prompt = strings.TrimSpace(fmt.Sprint(normalizedInput["prompt"])) + } + if prompt == "" { + return nil, errors.New("prompt is required") + } + taskType := strings.TrimSpace(req.Type) + if taskType == "" { + taskType = "text" + } + task := model.Task{ + ID: newID(), UserID: userID, SessionID: req.SessionID, ProjectID: req.ProjectID, + Type: taskType, Status: model.TaskStatusTextReplay, Stage: "文本持久化(前端自管)", Progress: 5, + Prompt: prompt, Operation: req.Operation, Provider: req.Provider, Model: strings.TrimSpace(req.Model), + } + if err := s.protectTaskSecrets(normalizedInput); err != nil { + return nil, err + } + inputJSON, _ := json.Marshal(normalizedInput) + task.InputJSON = string(inputJSON) + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + if err := s.createTaskWithinStorageQuota(&task, nil, policy); err != nil { + return nil, err + } + _ = s.log(userID, task.ID, "info", "文本持久化任务已创建(前端自管)", "") + return taskForOutput(task), nil +} + func (s *Service) requireCustomChannelsForTaskInput(input map[string]any) error { if !taskInputUsesCustomChannel(input) { return nil diff --git a/backend/internal/service/text_replay.go b/backend/internal/service/text_replay.go index 5db55053a..d1a82a3da 100644 --- a/backend/internal/service/text_replay.go +++ b/backend/internal/service/text_replay.go @@ -29,6 +29,49 @@ type TextReplayResult struct { Complete bool `json:"complete"` } +// isTextReplayTaskRequest 识别前端自管的文本持久化请求(input 带 replay=true)。 +// 这类任务由前端直连模型流式生成,仅作正文的后端存档与回放,不经过 worker 队列生成。 +func isTextReplayTaskRequest(input map[string]any) bool { + value, ok := input["replay"] + if !ok { + return false + } + switch v := value.(type) { + case bool: + return v + case string: + return strings.EqualFold(strings.TrimSpace(v), "true") + default: + return false + } +} + +// CompleteTextReplayTask 在文本生成结束后把最终正文写入任务,收尾为 succeeded, +// 关闭 delta 写入窗口并触发 text-deltas 归并,使 queryTaskTextReplay 能读到 finalText。 +func (s *Service) CompleteTextReplayTask(userID string, taskID string, text string) (*model.Task, error) { + if strings.TrimSpace(text) == "" { + return nil, BadAuthRequest("文本内容不能为空") + } + if len([]byte(text)) > textReplayMaxTaskBytes { + return nil, BadAuthRequest("文本内容过大") + } + if _, err := s.repo.TaskForUser(userID, taskID); err != nil { + return nil, err + } + resultJSON, _ := json.Marshal(map[string]any{"mode": "text", "text": text}) + completed, err := s.repo.CompleteTextReplayTask(userID, taskID, string(resultJSON), time.Now()) + if err != nil { + return nil, err + } + if !completed { + return nil, BadAuthRequest("该文本任务已结束或不属于你,无法完成") + } + if compactErr := s.finalizeTaskTextReplay(taskID, model.TaskStatusSucceeded); compactErr != nil { + _ = s.log(userID, taskID, "error", "文本回放窗口更新失败", compactErr.Error()) + } + return s.repo.TaskForUser(userID, taskID) +} + func (s *Service) AppendTaskTextDelta(userID string, taskID string, content string) (*model.TaskTextDelta, error) { task, err := s.repo.TaskForUser(userID, taskID) if err != nil { diff --git a/backend/internal/service/text_replay_test.go b/backend/internal/service/text_replay_test.go new file mode 100644 index 000000000..1608fcd99 --- /dev/null +++ b/backend/internal/service/text_replay_test.go @@ -0,0 +1,25 @@ +package service + +import "testing" + +func TestIsTextReplayTaskRequest(t *testing.T) { + cases := []struct { + name string + input map[string]any + want bool + }{ + {"replay 布尔 true", map[string]any{"replay": true}, true}, + {"replay 字符串 true", map[string]any{"replay": "true"}, true}, + {"replay 字符串 True 大小写", map[string]any{"replay": "TRUE"}, true}, + {"replay 布尔 false", map[string]any{"replay": false}, false}, + {"无 replay 字段", map[string]any{"mode": "text"}, false}, + {"replay 空字符串", map[string]any{"replay": ""}, false}, + {"replay 数字", map[string]any{"replay": 1}, false}, + {"replay nil", map[string]any{"replay": nil}, false}, + } + for _, tc := range cases { + if got := isTextReplayTaskRequest(tc.input); got != tc.want { + t.Errorf("%s: isTextReplayTaskRequest(%v) = %v, want %v", tc.name, tc.input, got, tc.want) + } + } +} diff --git a/web/src/lib/creation-text-replay.ts b/web/src/lib/creation-text-replay.ts new file mode 100644 index 000000000..04f97a3cb --- /dev/null +++ b/web/src/lib/creation-text-replay.ts @@ -0,0 +1,86 @@ +import { appendTaskTextDelta, completeTextReplayTask, createGenerationTask } from "@/services/api/task-center"; +import type { AiConfig } from "@/stores/use-config-store"; + +/** + * text-replay 前端发布器:为前端直连模型生成的文本,在后端维护一个 + * text_replay 任务作持久化存档 + 回放。 + * + * - start():创建后端 text_replay 任务(不排队、不计费),拿到 taskId; + * - publish(fullText):按阈值节流地把新增片段通过 appendTaskTextDelta 上报 + * (生成中途刷新页面也能读到已生成部分); + * - finish(finalText):把最终正文写入任务并置为 succeeded,queryTaskTextReplay + * 即可读到 finalText。 + * + * 全部失败静默降级:文本直连生成是主流程,后端持久化只是增强,失败不阻塞生成。 + */ +export function createTextReplayPublisher(config: AiConfig, prompt: string) { + let taskId: string | null = null; + let lastLen = 0; + let buffered = ""; + let flushTimer: ReturnType | null = null; + let closed = false; + + // 增量累积到该阈值才落一次库,避免逐 token 高频写。 + const FLUSH_THRESHOLD = 2048; + const flush = (force = false) => { + if (flushTimer) { + clearTimeout(flushTimer); + flushTimer = null; + } + if (closed || !taskId) return; + if (buffered.length >= FLUSH_THRESHOLD || force) { + const chunk = buffered; + buffered = ""; + if (chunk.trim()) void appendTaskTextDelta(taskId, chunk).catch(() => {}); + } else if (buffered) { + flushTimer = setTimeout(() => flush(true), 1000); + } + }; + + const start = async () => { + try { + const task = await createGenerationTask({ + type: "text", + prompt: prompt.trim() || "文本创作", + model: config.model, + input: { + mode: "text", + replay: true, + prompt: prompt.trim() || "文本创作", + config: { + model: config.model, + baseUrl: config.baseUrl, + apiKey: config.apiKey, + apiFormat: config.apiFormat, + }, + }, + }); + taskId = task?.id || null; + } catch { + taskId = null; // 降级:无持久化,不影响主流程 + } + }; + + const publish = (fullText: string) => { + if (!taskId) return; + if (fullText.length <= lastLen) return; + buffered += fullText.slice(lastLen); + lastLen = fullText.length; + flush(); + }; + + const finish = (finalText: string) => { + closed = true; + if (flushTimer) { + clearTimeout(flushTimer); + flushTimer = null; + } + if (!taskId) return; + void appendTaskTextDelta(taskId, buffered).catch(() => {}); + if (finalText && finalText.trim()) { + void completeTextReplayTask(taskId, finalText).catch(() => {}); + } + }; + + return { start, publish, finish }; +} diff --git a/web/src/pages/create/index.tsx b/web/src/pages/create/index.tsx index f5b2eb579..ee1a48498 100644 --- a/web/src/pages/create/index.tsx +++ b/web/src/pages/create/index.tsx @@ -24,6 +24,7 @@ import { isGenerationTaskCancelled, logicalModelIDForConfig, runBackendGeneratio import { requestImageQuestion, type AiTextContentPart } from "@/services/api/image"; import { listAddedSkills, type Skill } from "@/services/api/skills"; import { cancelGenerationTask, subscribeGenerationTasks, type GenerationTask } from "@/services/api/task-center"; +import { createTextReplayPublisher } from "@/lib/creation-text-replay"; import { isLocalDreaminaTaskId, isLocalDreaminaWaitStopped, localDreaminaCancellationCopy, localDreaminaCancellationMessage, localDreaminaDetachOutcome } from "@/services/local-dreamina-task-projection"; import { getMediaBlob, uploadMediaFile } from "@/services/file-storage"; import { uploadImage } from "@/services/image-storage"; @@ -536,10 +537,18 @@ export default function CreatePage() { ? await buildTextMessageContent(item) : item.content, }))); - await requestImageQuestion(requestConfig, history, (text) => updateOriginAssistant((item) => ({ ...item, content: text })), { + const replayPublisher = createTextReplayPublisher(requestConfig, text); + void replayPublisher.start(); + let finalText = ""; + await requestImageQuestion(requestConfig, history, (full) => { + finalText = full; + updateOriginAssistant((item) => ({ ...item, content: full })); + replayPublisher.publish(full); + }, { signal: requestLifecycle.signal, onReasoning: (reasoning) => updateOriginAssistant((item) => ({ ...item, reasoning })), }); + replayPublisher.finish(finalText); } } else if (mode === "image") { const taskCount = Math.max(1, Math.min(imageProfile.maxOutputs, Math.floor(Number(count) || 1))); diff --git a/web/src/services/api/task-center.ts b/web/src/services/api/task-center.ts index aaa0443f5..c39813b3e 100644 --- a/web/src/services/api/task-center.ts +++ b/web/src/services/api/task-center.ts @@ -391,6 +391,10 @@ export function appendTaskTextDelta(id: string, content: string) { return request(api.post(`/tasks/${encodeURIComponent(id)}/text-deltas`, { content })); } +export function completeTextReplayTask(id: string, text: string) { + return request(api.post(`/tasks/${encodeURIComponent(id)}/text-replay-complete`, { text })); +} + export function queryTaskTextReplay(id: string, after = 0) { return request(api.get(`/tasks/${encodeURIComponent(id)}/text-deltas`, { params: { after } })); }