feat(create): 创作对话文本模式接入 text-replay 持久化与回放

把后端既有但未接线的 text-replay 能力在前端打通:前端直连模型生成正文时,
把生成过程/结果持久化到后端任务,支持刷新重放与任务中心可见。

后端(最小配套,text-replay 本为前端自管设计):
- 新增 TaskStatusTextReplay 任务状态:不排入 worker 队列、不计费、不执行,
  仅作为正文增量(text-deltas)的存储容器;
- CreateTask 识别 input.replay=true 的请求,走 createTextReplayTask 轻量创建;
- 新增 POST /tasks/:id/text-replay-complete: 写入最终正文、置为 succeeded、
  触发 text-deltas 归并,使 queryTaskTextReplay 能读到 finalText。

前端:
- 新增 creation-text-replay.ts 发布器: start 建任务、publish 节流上报增量、
  finish 存最终全文,失败静默降级不阻塞生成;
- create 文本模式生成时接入发布器,正文流不经展示改动、额外持久化到后端。

测试: 新增 isTextReplayTaskRequest 纯函数测(本地)、text-replay 闭环仓测
(建任务后写增量 -> complete -> succeeded + finalText,CI/cgo 验证)。
build/vet/tsc 全绿。
This commit is contained in:
dlh
2026-08-19 21:45:42 +08:00
parent 298b5ebf1f
commit 35dee568c9
10 changed files with 336 additions and 1 deletions

View File

@@ -172,6 +172,26 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) {
} }
ok(c, task) 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) { r.GET("/tasks/:id/logs", func(c *gin.Context) {
user, err := currentUser(c, svc) user, err := currentUser(c, svc)
if err != nil { if err != nil {

View File

@@ -44,6 +44,10 @@ const (
TaskStatusSucceeded TaskStatus = "succeeded" TaskStatusSucceeded TaskStatus = "succeeded"
TaskStatusFailed TaskStatus = "failed" TaskStatusFailed TaskStatus = "failed"
TaskStatusCancelled TaskStatus = "cancelled" TaskStatusCancelled TaskStatus = "cancelled"
// text_replay 任务是前端自管、不经过 worker 队列的文本生成存档容器:
// 前端直连模型流式生成,把增量 POST 到 /tasks/:id/text-deltas 存档,
// 完成后调用 /tasks/:id/text-replay-complete 归并为最终正文。
TaskStatusTextReplay TaskStatus = "text_replay"
SessionStatusActive SessionStatus = "active" SessionStatusActive SessionStatus = "active"
SessionStatusCompleted SessionStatus = "completed" SessionStatusCompleted SessionStatus = "completed"

View File

@@ -124,6 +124,20 @@ func (r *Repository) CleanupTaskTextDeltas(now time.Time) (int64, error) {
return deleted, err 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 { func (r *Repository) DeleteUserTaskTextDeltas(userID string) error {
return r.db.Delete(&model.TaskTextDelta{}, "user_id = ?", userID).Error return r.db.Delete(&model.TaskTextDelta{}, "user_id = ?", userID).Error
} }

View File

@@ -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")
}
}

View File

@@ -347,6 +347,10 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task,
} }
normalizedInput = applyRoutedProviderSelection(normalizedInput, routed) normalizedInput = applyRoutedProviderSelection(normalizedInput, routed)
} }
// 前端自管的文本持久化任务:直连模型生成、增量上报 text-deltas,不排入 worker 队列生成。
if isTextReplayTaskRequest(normalizedInput) {
return s.createTextReplayTask(userID, req, normalizedInput)
}
if err := s.requireCustomChannelsForTaskInput(normalizedInput); err != nil { if err := s.requireCustomChannelsForTaskInput(normalizedInput); err != nil {
return nil, err return nil, err
} }
@@ -473,6 +477,41 @@ func normalizeTaskInput(input map[string]any) (map[string]any, error) {
return normalized, nil 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 { func (s *Service) requireCustomChannelsForTaskInput(input map[string]any) error {
if !taskInputUsesCustomChannel(input) { if !taskInputUsesCustomChannel(input) {
return nil return nil

View File

@@ -29,6 +29,49 @@ type TextReplayResult struct {
Complete bool `json:"complete"` 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) { func (s *Service) AppendTaskTextDelta(userID string, taskID string, content string) (*model.TaskTextDelta, error) {
task, err := s.repo.TaskForUser(userID, taskID) task, err := s.repo.TaskForUser(userID, taskID)
if err != nil { if err != nil {

View File

@@ -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)
}
}
}

View File

@@ -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<typeof setTimeout> | 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 };
}

View File

@@ -24,6 +24,7 @@ import { isGenerationTaskCancelled, logicalModelIDForConfig, runBackendGeneratio
import { requestImageQuestion, type AiTextContentPart } from "@/services/api/image"; import { requestImageQuestion, type AiTextContentPart } from "@/services/api/image";
import { listAddedSkills, type Skill } from "@/services/api/skills"; import { listAddedSkills, type Skill } from "@/services/api/skills";
import { cancelGenerationTask, subscribeGenerationTasks, type GenerationTask } from "@/services/api/task-center"; 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 { isLocalDreaminaTaskId, isLocalDreaminaWaitStopped, localDreaminaCancellationCopy, localDreaminaCancellationMessage, localDreaminaDetachOutcome } from "@/services/local-dreamina-task-projection";
import { getMediaBlob, uploadMediaFile } from "@/services/file-storage"; import { getMediaBlob, uploadMediaFile } from "@/services/file-storage";
import { uploadImage } from "@/services/image-storage"; import { uploadImage } from "@/services/image-storage";
@@ -536,10 +537,18 @@ export default function CreatePage() {
? await buildTextMessageContent(item) ? await buildTextMessageContent(item)
: item.content, : 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, signal: requestLifecycle.signal,
onReasoning: (reasoning) => updateOriginAssistant((item) => ({ ...item, reasoning })), onReasoning: (reasoning) => updateOriginAssistant((item) => ({ ...item, reasoning })),
}); });
replayPublisher.finish(finalText);
} }
} else if (mode === "image") { } else if (mode === "image") {
const taskCount = Math.max(1, Math.min(imageProfile.maxOutputs, Math.floor(Number(count) || 1))); const taskCount = Math.max(1, Math.min(imageProfile.maxOutputs, Math.floor(Number(count) || 1)));

View File

@@ -391,6 +391,10 @@ export function appendTaskTextDelta(id: string, content: string) {
return request<TaskTextDelta>(api.post(`/tasks/${encodeURIComponent(id)}/text-deltas`, { content })); return request<TaskTextDelta>(api.post(`/tasks/${encodeURIComponent(id)}/text-deltas`, { content }));
} }
export function completeTextReplayTask(id: string, text: string) {
return request<GenerationTask>(api.post(`/tasks/${encodeURIComponent(id)}/text-replay-complete`, { text }));
}
export function queryTaskTextReplay(id: string, after = 0) { export function queryTaskTextReplay(id: string, after = 0) {
return request<TaskTextReplay>(api.get(`/tasks/${encodeURIComponent(id)}/text-deltas`, { params: { after } })); return request<TaskTextReplay>(api.get(`/tasks/${encodeURIComponent(id)}/text-deltas`, { params: { after } }));
} }