diff --git a/backend/internal/database/schema.go b/backend/internal/database/schema.go index b6f1f3d55..b2f81598e 100644 --- a/backend/internal/database/schema.go +++ b/backend/internal/database/schema.go @@ -54,6 +54,7 @@ func Models() []any { &model.Announcement{}, &model.UserAnnouncementRead{}, &model.Task{}, + &model.TaskTextDelta{}, &model.Session{}, &model.Message{}, &model.TaskLog{}, diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index 5472bde9f..2c01aa0f0 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -11,6 +11,19 @@ import ( ) func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { + r.GET("/admin/text-replay-stats", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + stats, err := svc.AdminTextReplayStats(user) + if err != nil { + failService(c, err) + return + } + ok(c, stats) + }) r.POST("/tasks", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { @@ -65,6 +78,40 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, task) }) + r.POST("/tasks/:id/text-deltas", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + var req struct { + Content string `json:"content"` + } + if err := c.ShouldBindJSON(&req); err != nil { + fail(c, http.StatusBadRequest, err) + return + } + item, err := svc.AppendTaskTextDelta(user.ID, c.Param("id"), req.Content) + if err != nil { + failService(c, err) + return + } + ok(c, item) + }) + r.GET("/tasks/:id/text-deltas", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + after, _ := strconv.ParseInt(c.DefaultQuery("after", "0"), 10, 64) + result, err := svc.TaskTextReplay(user.ID, c.Param("id"), after) + if err != nil { + failService(c, err) + return + } + ok(c, result) + }) r.POST("/tasks/:id/retry", 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 8da363159..ff3e57b3f 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -754,6 +754,7 @@ type Task struct { LeaseExpiresAt *time.Time `json:"-" gorm:"index;index:idx_tasks_claim,priority:2"` InputJSON string `json:"inputJson" gorm:"type:text"` ResultJSON string `json:"resultJson" gorm:"type:text"` + TextDraft string `json:"textDraft,omitempty" gorm:"type:text"` Error string `json:"error"` Attempts int `json:"attempts"` StartedAt *time.Time `json:"startedAt"` @@ -762,6 +763,18 @@ type Task struct { UpdatedAt time.Time `json:"updatedAt"` } +// TaskTextDelta 只保存可回放窗口内的文本增量;最终正文和失败草稿分别归并到 Task.ResultJSON 与 Task.TextDraft。 +type TaskTextDelta struct { + ID string `json:"id" gorm:"primaryKey;size:36"` + UserID string `json:"userId" gorm:"index;size:36;index:idx_task_text_deltas_user_created,priority:1"` + TaskID string `json:"taskId" gorm:"index;size:36;uniqueIndex:idx_task_text_deltas_sequence,priority:1"` + Sequence int64 `json:"sequence" gorm:"uniqueIndex:idx_task_text_deltas_sequence,priority:2"` + Content string `json:"content" gorm:"type:text"` + ByteCount int64 `json:"byteCount"` + CreatedAt time.Time `json:"createdAt" gorm:"index:idx_task_text_deltas_user_created,priority:2"` + ExpiresAt time.Time `json:"expiresAt" gorm:"index"` +} + type Session struct { ID string `json:"id" gorm:"primaryKey;size:36"` UserID string `json:"userId" gorm:"index;size:36"` diff --git a/backend/internal/repository/admin_audit.go b/backend/internal/repository/admin_audit.go index 1b35c861f..075226328 100644 --- a/backend/internal/repository/admin_audit.go +++ b/backend/internal/repository/admin_audit.go @@ -58,6 +58,9 @@ func (r *Repository) BulkDisableUsers(actorID string, userIDs []string, events [ if err := tx.Delete(&model.AuthSession{}, "user_id IN ?", userIDs).Error; err != nil { return err } + if err := tx.Delete(&model.TaskTextDelta{}, "user_id IN ?", userIDs).Error; err != nil { + return err + } if err := tx.Model(&model.User{}).Where("id IN ?", userIDs).Updates(map[string]any{"status": model.UserStatusDisabled, "updated_at": now}).Error; err != nil { return err } diff --git a/backend/internal/repository/finance.go b/backend/internal/repository/finance.go index 6de9040be..1d039e88c 100644 --- a/backend/internal/repository/finance.go +++ b/backend/internal/repository/finance.go @@ -222,7 +222,7 @@ func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *m } updates := map[string]any{ "status": model.TaskStatusQueued, "stage": "等待队列调度", "progress": 5, "error": "", "result_json": "", - "started_at": nil, "completed_at": nil, "updated_at": time.Now(), + "text_draft": "", "started_at": nil, "completed_at": nil, "updated_at": time.Now(), } if order != nil { updates["billing_order_id"] = order.ID @@ -239,6 +239,9 @@ func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *m if updated.RowsAffected != 1 { return ErrTaskNotRetryable } + if err := tx.Delete(&model.TaskTextDelta{}, "user_id = ? AND task_id = ?", userID, taskID).Error; err != nil { + return err + } return tx.First(&task, "id = ? AND user_id = ?", taskID, userID).Error }) return &task, err diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index 4b8a4ddca..cddc02147 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -16,6 +16,10 @@ var ErrDailyUploadLimitExceeded = errors.New("daily upload limit exceeded") var ErrTaskProviderRecoveryConflict = errors.New("task provider recovery is already running") +var ErrTextReplayQuotaExceeded = errors.New("text replay quota exceeded") + +var ErrTextReplayClosed = errors.New("text replay task is closed") + type Repository struct { db *gorm.DB } @@ -52,9 +56,10 @@ func (r *Repository) UserStorageUsage(userID string) (UserStorageUsage, error) { (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(canvas_snapshot_json, '') AS BLOB)) + length(CAST(COALESCE(canvas_ops_json, '') AS BLOB))), 0) FROM sessions WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(content, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM messages WHERE user_id = ?) AS session_bytes, (SELECT COUNT(*) FROM tasks WHERE user_id = ?) AS task_count, - (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(input_json, '') AS BLOB)) + length(CAST(COALESCE(result_json, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB))), 0) FROM tasks WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(input_json, '') AS BLOB)) + length(CAST(COALESCE(result_json, '') AS BLOB)) + length(CAST(COALESCE(text_draft, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB))), 0) FROM tasks WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(message, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM task_logs WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(url, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM results WHERE user_id = ?) + + (SELECT COALESCE(SUM(byte_count), 0) FROM task_text_deltas WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(path, '') AS BLOB)) + length(CAST(COALESCE(model, '') AS BLOB)) + length(CAST(COALESCE(provider_request_id, '') AS BLOB)) + length(CAST(COALESCE(error_code, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB)) + length(CAST(COALESCE(upstream_url, '') AS BLOB)) + length(CAST(COALESCE(request_body, '') AS BLOB)) + length(CAST(COALESCE(response_body, '') AS BLOB))), 0) FROM api_call_logs WHERE user_id = ?) AS task_bytes, (SELECT COUNT(*) FROM api_call_logs WHERE user_id = ?) AS api_call_count ` @@ -62,7 +67,7 @@ func (r *Repository) UserStorageUsage(userID string) (UserStorageUsage, error) { query = strings.ReplaceAll(query, "length(CAST(COALESCE(", "octet_length(COALESCE(") query = strings.ReplaceAll(query, ", '') AS BLOB))", ", ''))") } - err := r.db.Raw(query, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID).Scan(&usage).Error + err := r.db.Raw(query, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID, userID).Scan(&usage).Error return usage, err } @@ -450,6 +455,15 @@ func (r *Repository) SessionForUser(userID string, id string) (*model.Session, e func (r *Repository) DeleteSessionDraft(userID string, id string) error { return r.db.Transaction(func(tx *gorm.DB) error { + var taskIDs []string + if err := tx.Model(&model.Task{}).Where("user_id = ? AND session_id = ?", userID, id).Pluck("id", &taskIDs).Error; err != nil { + return err + } + if len(taskIDs) > 0 { + if err := tx.Delete(&model.TaskTextDelta{}, "user_id = ? AND task_id IN ?", userID, taskIDs).Error; err != nil { + return err + } + } if err := tx.Delete(&model.Message{}, "user_id = ? AND session_id = ?", userID, id).Error; err != nil { return err } diff --git a/backend/internal/repository/text_replay.go b/backend/internal/repository/text_replay.go new file mode 100644 index 000000000..c93bcdf5b --- /dev/null +++ b/backend/internal/repository/text_replay.go @@ -0,0 +1,135 @@ +package repository + +import ( + "errors" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +type TextReplayLimits struct { + MaxTaskBytes int64 + MaxUserBytes int64 + MaxTaskEvents int64 +} + +type TextReplayStats struct { + EventCount int64 `json:"eventCount"` + TaskCount int64 `json:"taskCount"` + ByteCount int64 `json:"byteCount"` + OldestAt *time.Time `json:"oldestAt,omitempty"` +} + +func (r *Repository) AppendTaskTextDelta(userID string, taskID string, content string, expiresAt time.Time, limits TextReplayLimits) (*model.TaskTextDelta, error) { + var created model.TaskTextDelta + err := r.db.Transaction(func(tx *gorm.DB) error { + var task model.Task + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "user_id", "status").First(&task, "id = ? AND user_id = ?", taskID, userID).Error; err != nil { + return err + } + if task.Status == model.TaskStatusSucceeded || task.Status == model.TaskStatusFailed || task.Status == model.TaskStatusCancelled { + return ErrTextReplayClosed + } + var taskUsage struct { + Count int64 + Bytes int64 + Max int64 + } + if err := tx.Model(&model.TaskTextDelta{}).Select("count(*) AS count, COALESCE(sum(byte_count), 0) AS bytes, COALESCE(max(sequence), 0) AS max").Where("task_id = ?", taskID).Scan(&taskUsage).Error; err != nil { + return err + } + var userBytes int64 + if err := tx.Model(&model.TaskTextDelta{}).Select("COALESCE(sum(byte_count), 0)").Where("user_id = ?", userID).Scan(&userBytes).Error; err != nil { + return err + } + incoming := int64(len([]byte(content))) + if taskUsage.Count+1 > limits.MaxTaskEvents || taskUsage.Bytes+incoming > limits.MaxTaskBytes || userBytes+incoming > limits.MaxUserBytes { + return ErrTextReplayQuotaExceeded + } + created = model.TaskTextDelta{ID: newRepositoryID(), UserID: userID, TaskID: taskID, Sequence: taskUsage.Max + 1, Content: content, ByteCount: incoming, CreatedAt: time.Now(), ExpiresAt: expiresAt} + return tx.Create(&created).Error + }) + return &created, err +} + +func (r *Repository) TaskTextDeltas(userID string, taskID string, after int64, limit int) ([]model.TaskTextDelta, error) { + var items []model.TaskTextDelta + if limit <= 0 || limit > 1000 { + limit = 1000 + } + err := r.db.Where("user_id = ? AND task_id = ? AND sequence > ?", userID, taskID, after).Order("sequence asc").Limit(limit).Find(&items).Error + return items, err +} + +func (r *Repository) CompactTaskTextDeltas(taskID string, expiresAt time.Time, keepDraft bool) error { + return r.db.Transaction(func(tx *gorm.DB) error { + var task model.Task + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&task, "id = ?", taskID).Error; err != nil { + return err + } + if keepDraft { + var items []model.TaskTextDelta + if err := tx.Select("content").Where("task_id = ?", taskID).Order("sequence asc").Find(&items).Error; err != nil { + return err + } + if task.TextDraft == "" { + for _, item := range items { + task.TextDraft += item.Content + } + if err := tx.Model(&task).Update("text_draft", task.TextDraft).Error; err != nil { + return err + } + } + } + return tx.Model(&model.TaskTextDelta{}).Where("task_id = ?", taskID).Update("expires_at", expiresAt).Error + }) +} + +func (r *Repository) CleanupTaskTextDeltas(now time.Time) (int64, error) { + var deleted int64 + err := r.db.Transaction(func(tx *gorm.DB) error { + var taskIDs []string + if err := tx.Model(&model.TaskTextDelta{}).Where("expires_at <= ?", now).Distinct().Pluck("task_id", &taskIDs).Error; err != nil { + return err + } + for _, taskID := range taskIDs { + var task model.Task + if err := tx.Select("id", "status", "text_draft").First(&task, "id = ?", taskID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + continue + } + return err + } + if task.TextDraft != "" || (task.Status != model.TaskStatusFailed && task.Status != model.TaskStatusCancelled) { + continue + } + var items []model.TaskTextDelta + if err := tx.Select("content").Where("task_id = ?", taskID).Order("sequence asc").Find(&items).Error; err != nil { + return err + } + for _, item := range items { + task.TextDraft += item.Content + } + if err := tx.Model(&task).Update("text_draft", task.TextDraft).Error; err != nil { + return err + } + } + result := tx.Where("expires_at <= ? OR NOT EXISTS (SELECT 1 FROM tasks WHERE tasks.id = task_text_deltas.task_id)", now).Delete(&model.TaskTextDelta{}) + deleted = result.RowsAffected + return result.Error + }) + return deleted, err +} + +func (r *Repository) DeleteUserTaskTextDeltas(userID string) error { + return r.db.Delete(&model.TaskTextDelta{}, "user_id = ?", userID).Error +} + +func (r *Repository) TextReplayStats() (TextReplayStats, error) { + var stats TextReplayStats + err := r.db.Model(&model.TaskTextDelta{}).Select("count(*) AS event_count, count(DISTINCT task_id) AS task_count, COALESCE(sum(byte_count), 0) AS byte_count, min(created_at) AS oldest_at").Scan(&stats).Error + return stats, err +} diff --git a/backend/internal/service/admin.go b/backend/internal/service/admin.go index a6ced374a..1ba1a98e8 100644 --- a/backend/internal/service/admin.go +++ b/backend/internal/service/admin.go @@ -366,6 +366,9 @@ func (s *Service) DeleteUser(actor *model.User, userID string) error { if err := s.repo.DeleteUserAuthSessions(user.ID); err != nil { return err } + if err := s.repo.DeleteUserTaskTextDeltas(user.ID); err != nil { + return err + } // 有资金流水后必须保留用户主体,删除入口改为停用并清除全部登录态。 user.Status = model.UserStatusDisabled user.UpdatedAt = time.Now() diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index e31647b14..4c91a7566 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -189,6 +189,7 @@ func New(repo *repository.Repository, dataDir string) *Service { } func (s *Service) StartWorker() { + s.startTextReplayCleanup() go func() { slots := make(chan struct{}, maxChannelConcurrencyLimit) dispatch := func() { @@ -576,6 +577,9 @@ func (s *Service) CancelTask(userID string, id string) (*model.Task, error) { if task.SessionID != "" { _ = s.markSessionFailed(*task, "会话任务已取消。") } + if err := s.finalizeTaskTextReplay(task.ID, model.TaskStatusCancelled); err != nil { + _ = s.log(userID, task.ID, "error", "文本回放草稿归并失败", err.Error()) + } _ = s.log(userID, task.ID, "warn", "任务已取消", "") return taskForOutput(*task), nil } @@ -911,6 +915,9 @@ func (s *Service) processClaimedTask(task *model.Task) error { _ = s.MarkBillingUncertain(task.BillingOrderID, "任务取消时上游费用状态不明确") } _ = s.markSessionFailed(*task, "会话任务已取消。") + if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusCancelled); compactErr != nil { + _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) + } _ = s.log(task.UserID, task.ID, "warn", "任务已取消", "") return nil } @@ -922,6 +929,9 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) _ = s.repo.Save(task) + if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusFailed); compactErr != nil { + _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) + } if providerSucceeded || (!channelSlotFailedBeforeRequest && s.BillingFailureRequiresReview(task.BillingOrderID, task.ID, err)) { _ = s.MarkBillingUncertain(task.BillingOrderID, task.Error) } else { @@ -936,6 +946,9 @@ func (s *Service) processClaimedTask(task *model.Task) error { return err } if latest.Status == model.TaskStatusCancelled { + if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusCancelled); compactErr != nil { + _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) + } _ = s.MarkBillingUncertain(task.BillingOrderID, "上游已返回结果,但任务被取消") _ = s.markSessionFailed(*latest, "会话任务已取消。") _ = s.log(task.UserID, task.ID, "warn", "任务已取消,丢弃生成结果", "") @@ -952,11 +965,17 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) _ = s.repo.Save(task) + if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusFailed); compactErr != nil { + _ = s.log(task.UserID, task.ID, "error", "文本回放草稿归并失败", compactErr.Error()) + } _ = s.MarkBillingUncertain(task.BillingOrderID, "上游已成功但任务结果未保存:"+task.Error) _ = s.markSessionFailed(*task, task.Error) _ = s.log(task.UserID, task.ID, "error", "任务结果保存失败", task.Error) return err } + if compactErr := s.finalizeTaskTextReplay(task.ID, model.TaskStatusSucceeded); compactErr != nil { + _ = s.log(task.UserID, task.ID, "error", "文本回放窗口更新失败", compactErr.Error()) + } if completedTask, fetchErr := s.repo.Task(task.ID); fetchErr == nil { if registerErr := s.RegisterTaskOutputFromTask(*completedTask); registerErr != nil { // 任务成功与产物登记分开记账;登记失败保持步骤异常,允许项目页幂等补登记。 diff --git a/backend/internal/service/text_replay.go b/backend/internal/service/text_replay.go new file mode 100644 index 000000000..5db55053a --- /dev/null +++ b/backend/internal/service/text_replay.go @@ -0,0 +1,122 @@ +package service + +import ( + "encoding/json" + "errors" + "log" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" + + "gorm.io/gorm" +) + +const ( + textReplayMaxEventBytes = 64 << 10 + textReplayMaxTaskBytes = 2 << 20 + textReplayMaxUserBytes = 64 << 20 + textReplayMaxTaskEvents int64 = 4096 + textReplaySuccessRetention = 24 * time.Hour + textReplayDraftRetention = 7 * 24 * time.Hour +) + +type TextReplayResult struct { + Deltas []model.TaskTextDelta `json:"deltas"` + TextDraft string `json:"textDraft,omitempty"` + FinalText string `json:"finalText,omitempty"` + Complete bool `json:"complete"` +} + +func (s *Service) AppendTaskTextDelta(userID string, taskID string, content string) (*model.TaskTextDelta, error) { + task, err := s.repo.TaskForUser(userID, taskID) + if err != nil { + return nil, err + } + if capabilityFromTaskType(task.Type) != "text" { + return nil, BadAuthRequest("只有文本生成任务支持增量回放") + } + if strings.TrimSpace(content) == "" { + return nil, BadAuthRequest("文本增量不能为空") + } + if len([]byte(content)) > textReplayMaxEventBytes { + return nil, BadAuthRequest("单条文本增量不能超过 64KB") + } + item, err := s.repo.AppendTaskTextDelta(userID, taskID, content, time.Now().Add(textReplayDraftRetention), repository.TextReplayLimits{ + MaxTaskBytes: textReplayMaxTaskBytes, MaxUserBytes: textReplayMaxUserBytes, MaxTaskEvents: textReplayMaxTaskEvents, + }) + if errors.Is(err, repository.ErrTextReplayQuotaExceeded) { + return nil, BadAuthRequest("文本回放增量已达到配额,请等待任务归并后继续") + } + if errors.Is(err, repository.ErrTextReplayClosed) { + return nil, BadAuthRequest("已结束任务不能继续写入文本增量") + } + return item, err +} + +func (s *Service) TaskTextReplay(userID string, taskID string, after int64) (*TextReplayResult, error) { + task, err := s.repo.TaskForUser(userID, taskID) + if err != nil { + return nil, err + } + deltas, err := s.repo.TaskTextDeltas(userID, taskID, after, 1000) + if err != nil { + return nil, err + } + result := &TextReplayResult{Deltas: deltas, TextDraft: task.TextDraft, Complete: task.Status == model.TaskStatusSucceeded || task.Status == model.TaskStatusFailed || task.Status == model.TaskStatusCancelled} + if task.Status == model.TaskStatusSucceeded { + result.FinalText = taskResultText(task.ResultJSON) + } + return result, nil +} + +func (s *Service) finalizeTaskTextReplay(taskID string, status model.TaskStatus) error { + keepDraft := status == model.TaskStatusFailed || status == model.TaskStatusCancelled + retention := textReplaySuccessRetention + if keepDraft { + retention = textReplayDraftRetention + } + err := s.repo.CompactTaskTextDeltas(taskID, time.Now().Add(retention), keepDraft) + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + return err +} + +func (s *Service) CleanupTaskTextReplay() (int64, error) { + return s.repo.CleanupTaskTextDeltas(time.Now()) +} + +func (s *Service) AdminTextReplayStats(actor *model.User) (repository.TextReplayStats, error) { + if err := s.RequireAdmin(actor); err != nil { + return repository.TextReplayStats{}, err + } + return s.repo.TextReplayStats() +} + +func taskResultText(raw string) string { + var result struct { + Text string `json:"text"` + } + if json.Unmarshal([]byte(raw), &result) != nil { + return "" + } + return result.Text +} + +func (s *Service) startTextReplayCleanup() { + cleanup := func() { + if _, err := s.CleanupTaskTextReplay(); err != nil { + log.Printf("text replay cleanup failed: %v", err) + } + } + cleanup() + go func() { + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + for range ticker.C { + cleanup() + } + }() +} diff --git a/web/src/services/api/task-center.ts b/web/src/services/api/task-center.ts index 12ab01757..6bf5797af 100644 --- a/web/src/services/api/task-center.ts +++ b/web/src/services/api/task-center.ts @@ -30,6 +30,7 @@ export type GenerationTask = { previewKind?: "image" | "video"; inputJson?: string; resultJson?: string; + textDraft?: string; error?: string; attempts: number; startedAt?: string; @@ -57,6 +58,23 @@ export type ProviderTaskQueryResult = { billingSettled: boolean; }; +export type TaskTextDelta = { + id: string; + taskId: string; + sequence: number; + content: string; + byteCount: number; + createdAt: string; + expiresAt: string; +}; + +export type TaskTextReplay = { + deltas: TaskTextDelta[]; + textDraft?: string; + finalText?: string; + complete: boolean; +}; + export type AgentSession = { id: string; projectId?: string; @@ -199,6 +217,14 @@ export function queryGenerationTask(id: string, options?: { signal?: AbortSignal return request(api.get(`/tasks/${encodeURIComponent(id)}`, { signal: options?.signal })); } +export function appendTaskTextDelta(id: string, content: string) { + return request(api.post(`/tasks/${encodeURIComponent(id)}/text-deltas`, { content })); +} + +export function queryTaskTextReplay(id: string, after = 0) { + return request(api.get(`/tasks/${encodeURIComponent(id)}/text-deltas`, { params: { after } })); +} + export function retryGenerationTask(id: string) { return request(api.post(`/tasks/${encodeURIComponent(id)}/retry`)); }