diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index 981730b81..8a6071a7d 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -294,7 +294,7 @@ func (r *Repository) ClaimNextTask(owner string, leaseDuration time.Duration) (* now := time.Now() leaseExpiresAt := now.Add(leaseDuration) err := r.db.Transaction(func(tx *gorm.DB) error { - query := tx.Where("status = ? OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?))", model.TaskStatusQueued, model.TaskStatusRunning, now). + query := tx.Where("(status = ? OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?))) AND (next_poll_at IS NULL OR next_poll_at <= ?)", model.TaskStatusQueued, model.TaskStatusRunning, now, now). Order("created_at asc").Limit(1) if r.Dialect() == "postgres" { query = query.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}) @@ -309,7 +309,7 @@ func (r *Repository) ClaimNextTask(owner string, leaseDuration time.Duration) (* } claim := tx.Model(&model.Task{}).Where("id = ?", task.ID) if r.Dialect() != "postgres" { - claim = claim.Where("status = ? OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?))", model.TaskStatusQueued, model.TaskStatusRunning, now) + claim = claim.Where("(status = ? OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?))) AND (next_poll_at IS NULL OR next_poll_at <= ?)", model.TaskStatusQueued, model.TaskStatusRunning, now, now) } updated := claim. Updates(map[string]any{ @@ -320,6 +320,7 @@ func (r *Repository) ClaimNextTask(owner string, leaseDuration time.Duration) (* "started_at": gorm.Expr("COALESCE(started_at, ?)", now), "lease_owner": owner, "lease_expires_at": leaseExpiresAt, + "next_poll_at": nil, "updated_at": now, }) if updated.Error != nil { @@ -358,6 +359,23 @@ func (r *Repository) UpdateTaskProviderState(id string, providerRequestID string return r.db.Model(&model.Task{}).Where("id = ?", id).Updates(updates).Error } +func (r *Repository) DeferRunningTaskForProviderPoll(id string, owner string, stage string, delay time.Duration) error { + now := time.Now() + result := r.db.Model(&model.Task{}). + Where("id = ? AND status = ? AND lease_owner = ?", id, model.TaskStatusRunning, owner). + Updates(map[string]any{ + "stage": stage, "error": "", "completed_at": nil, "next_poll_at": now.Add(delay), + "lease_owner": "", "lease_expires_at": nil, "updated_at": now, + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskStateConflict + } + return nil +} + // 人工恢复仅锁定失败任务；旧 worker 的租约可覆盖，但未过期的人工恢复租约不能并发抢占。 func (r *Repository) ClaimFailedTaskProviderRecovery(id string, userID string, owner string, leaseDuration time.Duration) error { now := time.Now() diff --git a/backend/internal/repository/task_provider_poll_test.go b/backend/internal/repository/task_provider_poll_test.go new file mode 100644 index 000000000..76b0b84e8 --- /dev/null +++ b/backend/internal/repository/task_provider_poll_test.go @@ -0,0 +1,51 @@ +package repository + +import ( + "testing" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestDeferredProviderPollKeepsOriginalTaskIdentityWithoutImmediateReclaim(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:provider-poll-defer?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Task{}); err != nil { + t.Fatal(err) + } + task := model.Task{ + ID: "task-1", UserID: "user-1", Type: "canvas_video", Status: model.TaskStatusRunning, + LeaseOwner: "worker-1", LeaseExpiresAt: ptrTime(time.Now().Add(time.Minute)), ProviderRequestID: "provider-task-1", + } + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + repo := New(db) + if err := repo.DeferRunningTaskForProviderPoll(task.ID, task.LeaseOwner, "后台仍在生成", time.Minute); err != nil { + t.Fatal(err) + } + claimed, err := repo.ClaimNextTask("worker-2", 45*time.Second) + if err != nil { + t.Fatal(err) + } + if claimed != nil { + t.Fatalf("task reclaimed before next poll: %#v", claimed) + } + if err := db.Model(&model.Task{}).Where("id = ?", task.ID).Update("next_poll_at", time.Now().Add(-time.Second)).Error; err != nil { + t.Fatal(err) + } + claimed, err = repo.ClaimNextTask("worker-2", 45*time.Second) + if err != nil { + t.Fatal(err) + } + if claimed == nil || claimed.ID != task.ID || claimed.ProviderRequestID != task.ProviderRequestID { + t.Fatalf("reclaimed task = %#v", claimed) + } +} + +func ptrTime(value time.Time) *time.Time { return &value } diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index 46b9cd6cb..496192c16 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -1854,7 +1854,10 @@ func runNewAPIChannel2VideoTask(ctx context.Context, input canvasGenerationInput return nil, err } } - return nil, fmt.Errorf("NewAPI Video Generations 视频生成超时（任务 %s）", id) + if err := ctx.Err(); err != nil { + return nil, err + } + return nil, context.DeadlineExceeded } // 单次查询只读取既有上游任务，不创建新任务；自动轮询和人工恢复共用这条安全边界。 diff --git a/backend/internal/service/provider_test.go b/backend/internal/service/provider_test.go index 981843ea8..5dd6ee521 100644 --- a/backend/internal/service/provider_test.go +++ b/backend/internal/service/provider_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "io" "mime/multipart" "net/http" @@ -12,6 +13,9 @@ import ( "strconv" "strings" "testing" + "time" + + "infinite-canvas/backend/internal/model" ) const testReferenceImageDataURL = "data:image/png;base64,aGVsbG8=" @@ -1306,6 +1310,59 @@ func TestRunNewAPIChannel2VideoTaskDownloadsTemporaryResult(t *testing.T) { } } +func TestRunNewAPIChannel2VideoTaskResumesOriginalProviderTaskWithoutAnotherPost(t *testing.T) { + t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") + paths := make([]string, 0, 2) + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + paths = append(paths, r.Method+" "+r.URL.Path) + switch r.Method + " " + r.URL.Path { + case "GET /v1/video/generations/existing-provider-task": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"code":"success","data":{"task_id":"existing-provider-task","status":"SUCCESS","result_url":"` + server.URL + `/video.mp4"}}`)) + case "GET /video.mp4": + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + input := canvasGenerationInput{ + Mode: "video", + Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "video-model", InterfaceType: string(model.ChannelInterfaceNewAPIChannel2)}, + } + inputJSON, err := json.Marshal(input) + if err != nil { + t.Fatal(err) + } + ctx := withProviderAnalytics(context.Background(), nil, model.Task{ + ID: "task-1", Type: "canvas_video", ProviderRequestID: "existing-provider-task", InputJSON: string(inputJSON), + }) + result, err := runNewAPIChannel2VideoTask(ctx, input) + if err != nil { + t.Fatalf("runNewAPIChannel2VideoTask() error = %v", err) + } + if result["video"] == nil { + t.Fatalf("result = %#v", result) + } + want := "GET /v1/video/generations/existing-provider-task,GET /video.mp4" + if got := strings.Join(paths, ","); got != want { + t.Fatalf("paths = %q, want %q", got, want) + } +} + +func TestRunNewAPIChannel2VideoTaskReturnsTypedDeadlineWhenPollingWindowEnds(t *testing.T) { + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancel() + ctx = withProviderAnalytics(ctx, nil, model.Task{ID: "task-1", Type: "canvas_video", ProviderRequestID: "existing-provider-task"}) + _, err := runNewAPIChannel2VideoTask(ctx, canvasGenerationInput{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("runNewAPIChannel2VideoTask() error = %v, want context deadline exceeded", err) + } +} + func TestRunGeminiVeoVideoTaskUsesLongRunningOperation(t *testing.T) { t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") paths := make([]string, 0, 3) diff --git a/backend/internal/service/runtime_policy_test.go b/backend/internal/service/runtime_policy_test.go index ae34d3f0b..8a85de63a 100644 --- a/backend/internal/service/runtime_policy_test.go +++ b/backend/internal/service/runtime_policy_test.go @@ -1,7 +1,10 @@ package service import ( + "context" + "encoding/json" "testing" + "time" "infinite-canvas/backend/internal/model" "infinite-canvas/backend/internal/repository" @@ -10,6 +13,76 @@ import ( "gorm.io/gorm" ) +func TestVideoTaskTimeoutHasFiveMinuteSafetyFloor(t *testing.T) { + policy := defaultRuntimePolicy().Task + policy.VideoTimeoutMinutes = 1 + policy.ImageTimeoutMinutes = 1 + if got := taskExecutionTimeoutWithPolicy("canvas_video", policy); got != 5*time.Minute { + t.Fatalf("video timeout = %s, want 5m", got) + } + if got := taskExecutionTimeoutWithPolicy("canvas_image", policy); got != time.Minute { + t.Fatalf("image timeout = %s, want 1m", got) + } +} + +func TestOnlyResumableNewAPIChannel2VideoDeadlinesStayRunning(t *testing.T) { + svc := &Service{} + input, err := json.Marshal(canvasGenerationInput{Mode: "video", Config: providerConfig{BaseURL: "https://example.com", InterfaceType: string(model.ChannelInterfaceNewAPIChannel2)}}) + if err != nil { + t.Fatal(err) + } + base := model.Task{ID: "task-1", Type: "canvas_video", ProviderRequestID: "provider-task-1"} + if !svc.shouldDeferVideoProviderTask(base, string(input), context.DeadlineExceeded) { + t.Fatal("resumable NewAPI Channel 2 deadline should remain running") + } + if svc.shouldDeferVideoProviderTask(model.Task{ID: "task-2", Type: "canvas_video"}, string(input), context.DeadlineExceeded) { + t.Fatal("missing provider task id must not be deferred") + } + if svc.shouldDeferVideoProviderTask(base, string(input), context.Canceled) { + t.Fatal("explicit cancellation must not be deferred") + } + other, err := json.Marshal(canvasGenerationInput{Mode: "video", Config: providerConfig{BaseURL: "https://example.com", InterfaceType: string(model.ChannelInterfaceNewAPIVideo)}}) + if err != nil { + t.Fatal(err) + } + if svc.shouldDeferVideoProviderTask(base, string(other), context.DeadlineExceeded) { + t.Fatal("providers without a verified query-only resume contract must fail closed") + } +} + +func TestResumableVideoDeadlineUsesResolvedSystemChannelProtocol(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:"+newID()+"?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.ModelChannel{}, &model.ChannelModel{}); err != nil { + t.Fatal(err) + } + channel := model.ModelChannel{ + ID: "channel-video", UserID: "admin", Scope: model.ChannelScopeSystem, Enabled: true, Name: "Video", + BaseURL: "https://example.com", APIKey: "test-key", APIFormat: "openai", ModelsJSON: `["video-model"]`, + } + channelModel := model.ChannelModel{ + ID: "model-video", ChannelID: channel.ID, ModelKey: "video-model", Capability: "video", + Protocol: model.ChannelInterfaceNewAPIChannel2, Enabled: true, PriceConfigured: true, + } + if err := db.Create(&channel).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&channelModel).Error; err != nil { + t.Fatal(err) + } + input, err := json.Marshal(canvasGenerationInput{Mode: "video", Config: providerConfig{ChannelID: channel.ID, Model: "video-model"}}) + if err != nil { + t.Fatal(err) + } + svc := New(repository.New(db), t.TempDir()) + task := model.Task{ID: "task-system", Type: "canvas_video", ProviderRequestID: "provider-task-1"} + if !svc.shouldDeferVideoProviderTask(task, string(input), context.DeadlineExceeded) { + t.Fatal("system-channel task must use the resolved model protocol when deciding query-only recovery") + } +} + func TestRuntimePolicyDefaultsAndSelfUseModeValidate(t *testing.T) { if err := validateRuntimePolicy(defaultRuntimePolicy()); err != nil { t.Fatalf("default runtime policy error = %v", err) diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index dd93883c9..691ced9cc 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -990,6 +990,14 @@ func (s *Service) processClaimedTask(task *model.Task) error { return nil } if errors.Is(err, context.DeadlineExceeded) { + decryptedInput, decryptErr := s.decryptTaskInputJSON(task.InputJSON) + if decryptErr == nil && s.shouldDeferVideoProviderTask(*task, decryptedInput, err) { + if deferErr := s.repo.DeferRunningTaskForProviderPoll(task.ID, task.LeaseOwner, "后台仍在生成", 15*time.Second); deferErr != nil { + return deferErr + } + _ = s.log(task.UserID, task.ID, "info", "前台等待结束，上游视频仍在生成，将继续回查原任务", task.PollStage) + return nil + } err = errors.New(taskTimeoutMessage(task.Type)) } task.Status = model.TaskStatusFailed @@ -1079,7 +1087,7 @@ func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) t case taskType == "agent_storyboard" || taskType == "agent_storyboard_rows": return time.Duration(policy.StoryboardTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_"): - return time.Duration(policy.VideoTimeoutMinutes) * time.Minute + return max(time.Duration(policy.VideoTimeoutMinutes)*time.Minute, 5*time.Minute) case strings.HasPrefix(taskType, "canvas_image"): return time.Duration(policy.ImageTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_audio"): @@ -1091,6 +1099,18 @@ func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) t } } +func (s *Service) shouldDeferVideoProviderTask(task model.Task, decryptedInput string, err error) bool { + if !errors.Is(err, context.DeadlineExceeded) || strings.TrimSpace(task.ProviderRequestID) == "" || (!strings.HasPrefix(task.Type, "canvas_video") && !strings.HasPrefix(task.Type, "video_")) { + return false + } + var input canvasGenerationInput + if json.Unmarshal([]byte(decryptedInput), &input) != nil { + return false + } + resolved, resolveErr := s.resolveProviderConfig(input.Config) + return resolveErr == nil && resolved.InterfaceType == string(model.ChannelInterfaceNewAPIChannel2) +} + func taskTimeoutMessage(taskType string) string { if strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_") { return "视频生成等待超时，请稍后到任务中心查看或重试。" diff --git a/web/src/components/canvas/canvas-local-agent-panel.tsx b/web/src/components/canvas/canvas-local-agent-panel.tsx index 0693c5eea..7d42da2f4 100644 --- a/web/src/components/canvas/canvas-local-agent-panel.tsx +++ b/web/src/components/canvas/canvas-local-agent-panel.tsx @@ -10,7 +10,18 @@ import { createClientId } from "@/lib/client-id"; import { getLocalRuntimeSessionClient, useLocalRuntimeStore } from "@/stores/use-local-runtime-store"; import { useThemeStore } from "@/stores/use-theme-store"; import { useUserStore } from "@/stores/use-user-store"; -import { useCanvasAgentStore, type AgentAttachment, type AgentChatItem, type AgentEventLog, type AgentPanelTab, type AgentPendingToolCall, type AgentThreadSummary } from "@/stores/canvas/use-canvas-agent-store"; +import { + canvasAgentConnectionStatusText, + canvasAgentConnectionStartingPatch, + canvasAgentTransientDisconnectPatch, + useCanvasAgentStore, + type AgentAttachment, + type AgentChatItem, + type AgentEventLog, + type AgentPanelTab, + type AgentPendingToolCall, + type AgentThreadSummary, +} from "@/stores/canvas/use-canvas-agent-store"; import { previewCanvasAgentOps, summarizeCanvasAgentOps, type CanvasAgentOp, type CanvasAgentSnapshot } from "@/lib/canvas/canvas-agent-ops"; import { isProjectAgentReadTool, isProjectAgentToolName, runProjectAgentTool } from "@/services/api/project-agent-tools"; import { AgentChatComposer, AgentChatMessage, AgentPanelTabs, AgentPendingToolCard, AgentWorkingMessage, type CanvasAgentChatAttachment } from "./canvas-agent-chat-ui"; @@ -236,7 +247,7 @@ export const CanvasLocalAgentPanel = memo(function CanvasLocalAgentPanel({ } errorLoggedRef.current = true; connectedRef.current = false; - clearAgentSession({ activity: wasConnected ? "正在重连" : "连接失败", connected: false, connectError: text }); + setAgentState(canvasAgentTransientDisconnectPatch(wasConnected ? "正在重连" : "连接失败", text)); await waitForCanvasRuntimeReconnect(controller.signal); } } @@ -402,28 +413,15 @@ export const CanvasLocalAgentPanel = memo(function CanvasLocalAgentPanel({ if (connected) syncState(clientIdRef.current, restored); }; - const toggleAgentConnection = async () => { + const toggleAgentConnection = () => { if (enabled) { connectionControllerRef.current?.abort(); - clearAgentSession({ enabled: false, connected: false, activity: "离线", connectError: "" }); + pendingToolRef.current = null; + setAgentState({ enabled: false, connected: false, activity: "离线", connectError: "", waiting: false, sending: false, pendingTool: null }); return; } - const controller = new AbortController(); - connectionControllerRef.current = controller; - setAgentState({ connected: false, activity: "连接中", connectError: "", activeTab: "setup" }); - try { - await prepareCanvasRuntimeConnection(useLocalRuntimeStore, controller.signal); - if (controller.signal.aborted) return; - errorLoggedRef.current = false; - setAgentState({ enabled: true, connected: false, activity: "连接中", connectError: "", activeTab: "setup" }); - } catch (error) { - if (controller.signal.aborted) return; - const text = error instanceof Error ? error.message : "本机 Runtime 连接失败"; - setAgentState({ connectError: text }); - if (!headless) message.warning(text); - } finally { - if (connectionControllerRef.current === controller) connectionControllerRef.current = null; - } + errorLoggedRef.current = false; + setAgentState(canvasAgentConnectionStartingPatch()); }; useEffect(() => { @@ -432,21 +430,6 @@ export const CanvasLocalAgentPanel = memo(function CanvasLocalAgentPanel({ void toggleAgentConnection(); }, [autoConnect, connected, enabled]); - function clearAgentSession(patch: Parameters[0] = {}) { - setAgentState({ - messages: [], - threads: [], - activeThreadId: "", - workspacePath: "", - loadingThreads: false, - waiting: false, - sending: false, - pendingTool: null, - ...patch, - }); - pendingToolRef.current = null; - } - const startNewThread = async () => { const projectId = snapshotRef.current.projectId; if (!connected || !projectId) return; @@ -786,8 +769,8 @@ function AgentConnectView({ connectError: string; onToggleEnabled: () => void; }) { - const statusText = connectError ? "连接失败" : connected ? activity : enabled ? "连接中" : "未连接"; - const statusColor = connectError ? "#dc2626" : connected ? "#16a34a" : enabled ? "#d97706" : theme.node.muted; + const statusText = canvasAgentConnectionStatusText({ enabled, connected, activity, connectError }); + const statusColor = statusText === "连接失败" ? "#dc2626" : connected ? "#16a34a" : enabled ? "#d97706" : theme.node.muted; return (
diff --git a/web/src/services/api/generation-task.ts b/web/src/services/api/generation-task.ts index eb4b08463..b2372c560 100644 --- a/web/src/services/api/generation-task.ts +++ b/web/src/services/api/generation-task.ts @@ -7,6 +7,7 @@ import { isLocalDreaminaBackgroundTask, localDreaminaTaskId, projectLocalDreamin import { modelCapabilityConfigFor } from "@/lib/model-capabilities"; import { grokImagePromptLimitError } from "@/lib/grok-image-prompt-limit"; import { resolveModelRequestConfig, type AiConfig } from "@/stores/use-config-store"; +import { useLocalDreaminaModelStore } from "@/stores/use-local-dreamina-model-store"; import type { ReferenceImage } from "@/types/image"; import type { ReferenceAudio, ReferenceVideo } from "@/types/media"; @@ -46,6 +47,7 @@ export type GenerationTaskDependencies = { runLocal: (input: LocalDreaminaGenerationInput, signal?: AbortSignal, onTaskUpdate?: (task: LocalDreaminaGenerationTask) => void) => ReturnType; createId: () => string; now: () => string; + ensureLocalDreaminaReady?: (signal?: AbortSignal) => Promise; }; const defaultDependencies: GenerationTaskDependencies = { @@ -54,6 +56,7 @@ const defaultDependencies: GenerationTaskDependencies = { runLocal: (input, signal, onTaskUpdate) => runLocalDreaminaGenerationTask(input, { onTaskUpdate }, signal), createId: () => crypto.randomUUID(), now: () => new Date().toISOString(), + ensureLocalDreaminaReady: (signal) => useLocalDreaminaModelStore.getState().ensureReady(signal), }; type PreparedGenerationReferences = { @@ -88,6 +91,8 @@ export async function runBackendGenerationTask( throwIfAborted(signal); assertClientPromptLimit(mode, prompt, config, metadata); if (isLocalDreaminaModel(config.model)) { + await dependencies.ensureLocalDreaminaReady?.(signal); + throwIfAborted(signal); return await runLocalDreaminaGeneration( { projectId, mode, prompt, config, referenceImages, referenceVideos, referenceAudios, mask, signal, metadata, onTaskUpdate, localIdempotencyKey, localResumeOnly, clientOperationId, retryOf, attemptGroupId }, dependencies, @@ -104,6 +109,8 @@ export async function runBackendGenerationTaskBatch(options: BackendGenerationTa assertClientPromptLimit(options.mode, options.prompt, options.config, options.metadata); if (options.retryContextsByBatchIndex && options.retryContextsByBatchIndex.length !== count) throw new Error("生成重试批次任务数量不匹配"); if (isLocalDreaminaModel(options.config.model)) { + await dependencies.ensureLocalDreaminaReady?.(options.signal); + throwIfAborted(options.signal); return Promise.allSettled( Array.from({ length: count }, (_, batchIndex) => { const retryContext = options.retryContextsByBatchIndex?.[batchIndex]; @@ -191,7 +198,8 @@ async function runLocalDreaminaGeneration(options: BackendGenerationTaskOptions, }, ); const completedAt = dependencies.now(); - options.onTaskUpdate?.({ ...task, status: "succeeded", progress: 100, stage: "local_cli_succeeded", resultJson: JSON.stringify(result), completedAt, updatedAt: completedAt }); + latestPublicTask = { ...latestPublicTask, status: "succeeded", progress: 100, stage: "local_cli_succeeded", resultJson: JSON.stringify(result), completedAt, updatedAt: completedAt }; + options.onTaskUpdate?.(latestPublicTask); return result; } catch (error) { const completedAt = dependencies.now(); @@ -200,7 +208,7 @@ async function runLocalDreaminaGeneration(options: BackendGenerationTaskOptions, const localErrorCode = error instanceof LocalDreaminaGenerationClientError ? error.code : undefined; if (!(cancelled && isLocalDreaminaBackgroundTask(latestPublicTask))) { options.onTaskUpdate?.({ - ...task, + ...latestPublicTask, status: cancelled ? "cancelled" : "failed", stage: cancelled ? "local_cli_cancelled" : "local_cli_failed", completedAt, diff --git a/web/src/stores/canvas/use-canvas-agent-store.ts b/web/src/stores/canvas/use-canvas-agent-store.ts index 88e4984e8..b9f8a0be2 100644 --- a/web/src/stores/canvas/use-canvas-agent-store.ts +++ b/web/src/stores/canvas/use-canvas-agent-store.ts @@ -35,10 +35,50 @@ type CanvasAgentStore = { clearEventLogs: () => void; }; +const CANVAS_AGENT_ENABLED_STORAGE_KEY = "canvas-agent-enabled"; + +type CanvasAgentPreferenceStorage = { + getItem(key: string): string | null; + setItem(key: string, value: string): void; +}; + +export function readCanvasAgentEnabledPreference(storage: Pick | undefined = browserPreferenceStorage()) { + try { + return storage?.getItem(CANVAS_AGENT_ENABLED_STORAGE_KEY) === "true"; + } catch { + return false; + } +} + +export function writeCanvasAgentEnabledPreference(enabled: boolean, storage: Pick | undefined = browserPreferenceStorage()) { + try { + storage?.setItem(CANVAS_AGENT_ENABLED_STORAGE_KEY, String(enabled)); + } catch { + // 浏览器隐私模式可能拒绝本地偏好写入；连接本身仍可继续。 + } +} + +export function canvasAgentConnectionStartingPatch() { + return { enabled: true, connected: false, activity: "连接中", connectError: "", activeTab: "setup" as const }; +} + +export function canvasAgentTransientDisconnectPatch(activity: string, connectError: string) { + return { enabled: true, connected: false, activity, connectError, waiting: false, sending: false }; +} + +export function canvasAgentConnectionStatusText({ enabled, connected, activity, connectError }: Pick) { + if (enabled && !connected && activity === "正在重连") return activity; + return connectError ? "连接失败" : connected ? activity : enabled ? "连接中" : "未连接"; +} + +function browserPreferenceStorage(): CanvasAgentPreferenceStorage | undefined { + return typeof window === "undefined" ? undefined : window.localStorage; +} + export const useCanvasAgentStore = create((set) => ({ width: typeof window === "undefined" ? 440 : Number(localStorage.getItem("canvas-agent-panel-width")) || 440, connected: false, - enabled: false, + enabled: readCanvasAgentEnabledPreference(), prompt: "", attachments: [], sending: false, @@ -54,7 +94,10 @@ export const useCanvasAgentStore = create((set) => ({ activity: "就绪", connectError: "", pendingTool: null, - setAgentState: (patch) => set(patch), + setAgentState: (patch) => { + if (typeof patch.enabled === "boolean") writeCanvasAgentEnabledPreference(patch.enabled); + set(patch); + }, addMessage: (item) => set((state) => ({ messages: [...state.messages.slice(-120), item] })), addEventLog: (item) => set((state) => ({ eventLogs: [...state.eventLogs.slice(-160), item] })), clearEventLogs: () => set({ eventLogs: [] }), diff --git a/web/src/stores/use-local-dreamina-model-store.ts b/web/src/stores/use-local-dreamina-model-store.ts index c19dbcfce..18b1b62bf 100644 --- a/web/src/stores/use-local-dreamina-model-store.ts +++ b/web/src/stores/use-local-dreamina-model-store.ts @@ -4,31 +4,103 @@ import { create } from "zustand"; import { getDreaminaModelCatalogSnapshotWithSessionRecovery, type DreaminaLocalModel } from "@/services/local-dreamina-model-catalog"; import { getLocalRuntimeSessionClient, useLocalRuntimeStore } from "@/stores/use-local-runtime-store"; -type State = { state: "idle" | "loading" | "ready" | "error"; models: DreaminaLocalModel[]; cacheScope?: string; sync(signal?: AbortSignal): Promise }; +type CatalogSnapshot = Awaited>; +type RuntimeCatalogState = { connection: string; modules: Array<{ id: string; scopes: string[] }> }; +type Dependencies = { + getRuntimeState(): RuntimeCatalogState; + getClient(): Parameters[0]; + loadSnapshot(client: Parameters[0], signal: AbortSignal): Promise; + requestTimeoutMs?: number; +}; +type State = { + state: "idle" | "loading" | "ready" | "error"; + models: DreaminaLocalModel[]; + cacheScope?: string; + sync(signal?: AbortSignal): Promise; + ensureReady(signal?: AbortSignal): Promise; +}; export const dreaminaModelCacheScopeKey = ({ accountBinding, sessionEpoch }: { accountBinding: string; sessionEpoch: string | number }) => `${accountBinding}:${sessionEpoch}`; -export const useLocalDreaminaModelStore = create((set, get) => ({ - state: "idle", - models: [], - cacheScope: undefined, - async sync(signal) { - const runtime = useLocalRuntimeStore.getState(); - const available = runtime.connection === "connected" && runtime.modules.some((module) => module.id === "dreamina" && module.scopes.includes("dreamina:models")); - if (!available) { - set({ state: "idle", models: [], cacheScope: undefined }); - return; - } - const client = getLocalRuntimeSessionClient(); - try { - const snapshot = await getDreaminaModelCatalogSnapshotWithSessionRecovery(client, signal); - const cacheScope = dreaminaModelCacheScopeKey(snapshot); - set({ state: "ready", models: snapshot.models, cacheScope }); - } catch { - if (!signal?.aborted) set({ state: "error", models: [], cacheScope: undefined }); - } - }, -})); +const DEFAULT_CATALOG_TIMEOUT_MS = 8_000; + +export function createLocalDreaminaModelStore(dependencies: Dependencies) { + let requestRevision = 0; + let activeRequest: { revision: number; controller: AbortController; promise: Promise } | null = null; + return create((set, get) => { + const load = (force: boolean, signal?: AbortSignal) => { + const runtime = dependencies.getRuntimeState(); + const available = runtime.connection === "connected" && runtime.modules.some((module) => module.id === "dreamina" && module.scopes.includes("dreamina:models")); + if (!available) { + requestRevision++; + activeRequest?.controller.abort(); + activeRequest = null; + set({ state: "idle", models: [], cacheScope: undefined }); + return Promise.reject(new Error("即梦本机模型目录尚未就绪")); + } + const current = get(); + if (!force && current.state === "ready" && current.models.length) return Promise.resolve(current.models); + if (force && activeRequest) { + requestRevision++; + activeRequest.controller.abort(); + activeRequest = null; + } + if (!activeRequest) { + set({ state: "loading" }); + const revision = ++requestRevision; + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), Math.max(1, dependencies.requestTimeoutMs ?? DEFAULT_CATALOG_TIMEOUT_MS)); + const promise = dependencies + .loadSnapshot(dependencies.getClient(), controller.signal) + .then((snapshot) => { + if (revision === requestRevision) set({ state: "ready", models: snapshot.models, cacheScope: dreaminaModelCacheScopeKey(snapshot) }); + return snapshot.models; + }) + .catch((error) => { + if (revision === requestRevision) set({ state: "error", models: [], cacheScope: undefined }); + throw error; + }) + .finally(() => { + clearTimeout(timer); + if (activeRequest?.revision === revision) activeRequest = null; + }); + activeRequest = { revision, controller, promise }; + } + return waitForCatalog(activeRequest.promise, signal); + }; + return { + state: "idle", + models: [], + cacheScope: undefined, + async sync(signal) { + try { + await load(true, signal); + } catch { + // 页面引导读取失败只投影 store 状态；生成入口的 ensureReady 会明确失败。 + } + }, + ensureReady(signal) { + return load(false, signal); + }, + }; + }); +} + +function waitForCatalog(promise: Promise, signal?: AbortSignal): Promise { + if (!signal) return promise; + if (signal.aborted) return Promise.reject(new DOMException("Aborted", "AbortError")); + return new Promise((resolve, reject) => { + const abort = () => reject(new DOMException("Aborted", "AbortError")); + signal.addEventListener("abort", abort, { once: true }); + promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", abort)); + }); +} + +export const useLocalDreaminaModelStore = createLocalDreaminaModelStore({ + getRuntimeState: () => useLocalRuntimeStore.getState(), + getClient: getLocalRuntimeSessionClient, + loadSnapshot: (client, signal) => getDreaminaModelCatalogSnapshotWithSessionRecovery(client, signal), +}); export function useLocalDreaminaModelBootstrap() { const connection = useLocalRuntimeStore((state) => state.connection); diff --git a/web/test/canvas-local-runtime.test.ts b/web/test/canvas-local-runtime.test.ts index c504393d4..898b6951d 100644 --- a/web/test/canvas-local-runtime.test.ts +++ b/web/test/canvas-local-runtime.test.ts @@ -98,6 +98,73 @@ test("Canvas Agent UI store contains no Runtime endpoint, bearer, or session fie expect("session" in state).toBe(false); }); +test("Canvas Agent reconnect intent persists while transient disconnects preserve the active conversation", async () => { + const module = await import("../src/stores/canvas/use-canvas-agent-store").catch(() => ({})); + const connectionStartingPatch = ( + module as { + canvasAgentConnectionStartingPatch?: () => Record; + } + ).canvasAgentConnectionStartingPatch; + const transientDisconnectPatch = ( + module as { + canvasAgentTransientDisconnectPatch?: (activity: string, connectError: string) => Record; + } + ).canvasAgentTransientDisconnectPatch; + const writeEnabled = ( + module as { + writeCanvasAgentEnabledPreference?: (enabled: boolean, storage: { setItem(key: string, value: string): void }) => void; + } + ).writeCanvasAgentEnabledPreference; + const readEnabled = ( + module as { + readCanvasAgentEnabledPreference?: (storage: { getItem(key: string): string | null }) => boolean; + } + ).readCanvasAgentEnabledPreference; + + expect(typeof connectionStartingPatch).toBe("function"); + expect(typeof transientDisconnectPatch).toBe("function"); + expect(typeof writeEnabled).toBe("function"); + expect(typeof readEnabled).toBe("function"); + if (!connectionStartingPatch || !transientDisconnectPatch || !writeEnabled || !readEnabled) return; + + const durable = { + messages: [{ id: "message-1", role: "assistant", text: "kept" }], + threads: [{ id: "thread-1", preview: "kept" }], + activeThreadId: "thread-1", + workspacePath: "D:/workspace", + pendingTool: { requestId: "tool-1", name: "canvas_get_state" }, + }; + const starting = { ...durable, ...connectionStartingPatch() }; + expect(starting).toMatchObject({ enabled: true, connected: false, activity: "连接中", ...durable }); + + const reconnecting = { ...starting, ...transientDisconnectPatch("正在重连", "连接已断开") }; + expect(reconnecting).toMatchObject({ enabled: true, connected: false, activity: "正在重连", connectError: "连接已断开", ...durable }); + + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + }; + writeEnabled(true, storage); + expect(readEnabled(storage)).toBe(true); + writeEnabled(false, storage); + expect(readEnabled(storage)).toBe(false); +}); + +test("Canvas Agent status shows reconnecting instead of a stale connection failure", async () => { + const module = await import("../src/stores/canvas/use-canvas-agent-store").catch(() => ({})); + const statusText = ( + module as { + canvasAgentConnectionStatusText?: (state: { enabled: boolean; connected: boolean; activity: string; connectError: string }) => string; + } + ).canvasAgentConnectionStatusText; + expect(typeof statusText).toBe("function"); + if (!statusText) return; + + expect(statusText({ enabled: true, connected: false, activity: "正在重连", connectError: "连接已断开" })).toBe("正在重连"); + expect(statusText({ enabled: true, connected: false, activity: "连接失败", connectError: "首次连接失败" })).toBe("连接失败"); +}); + test("Canvas connection reuses the shared Runtime store and requires the canvas module", async () => { const module = await import("../src/lib/canvas/local-runtime-connection"); const prepare = ( diff --git a/web/test/generation-task-local.test.ts b/web/test/generation-task-local.test.ts index 3b1ef7662..0fc2b12b1 100644 --- a/web/test/generation-task-local.test.ts +++ b/web/test/generation-task-local.test.ts @@ -335,6 +335,140 @@ test("shared Create and Canvas task projection exposes queued, submitted, genera expect(distinctUpdates[4]?.resultJson).toContain('"mode":"video"'); }); +test("the first local Dreamina submission waits for catalog readiness before invoking the paid Runtime path", async () => { + const order: string[] = []; + let release!: () => void; + const readyGate = new Promise((resolve) => { + release = resolve; + }); + const pending = runBackendGenerationTask( + { + mode: "video", + prompt: "first visit", + config: { ...defaultConfig, model: "local:dreamina-cli:seedance2.0mini", videoSeconds: "4", vquality: "720" }, + }, + { + createTask: async () => { + throw new Error("must not create backend task"); + }, + waitTask: async () => { + throw new Error("must not wait backend task"); + }, + ensureLocalDreaminaReady: async () => { + order.push("catalog-start"); + await readyGate; + order.push("catalog-ready"); + }, + runLocal: async () => { + order.push("paid-runtime"); + return { mode: "video", video: { dataUrl: "data:video/mp4;base64,AAAA", mimeType: "video/mp4", bytes: 3 } }; + }, + createId: () => "dreamina-first-ready-0001", + now: () => "2026-08-18T00:00:00.000Z", + }, + ); + + await Promise.resolve(); + expect(order).toEqual(["catalog-start"]); + release(); + await pending; + expect(order).toEqual(["catalog-start", "catalog-ready", "paid-runtime"]); +}); + +test("the first local Dreamina batch shares one catalog readiness gate before any paid Runtime call", async () => { + const order: string[] = []; + let release!: () => void; + const readyGate = new Promise((resolve) => { + release = resolve; + }); + const pending = runBackendGenerationTaskBatch( + { + mode: "image", + prompt: "first batch", + config: { ...defaultConfig, model: "local:dreamina-cli:5.0Pro", count: "1" }, + count: 3, + }, + { + createTask: async () => { + throw new Error("must not create backend task"); + }, + waitTask: async () => { + throw new Error("must not wait backend task"); + }, + ensureLocalDreaminaReady: async () => { + order.push("catalog-start"); + await readyGate; + order.push("catalog-ready"); + }, + runLocal: async () => { + order.push("paid-runtime"); + return { mode: "image", images: [{ dataUrl: "data:image/png;base64,AAAA", mimeType: "image/png", bytes: 3 }] }; + }, + createId: () => `dreamina-first-batch-${order.length}`, + now: () => "2026-08-18T00:00:00.000Z", + }, + ); + + await Promise.resolve(); + expect(order).toEqual(["catalog-start"]); + release(); + const settled = await pending; + expect(settled.every((entry) => entry.status === "fulfilled")).toBe(true); + expect(order).toEqual(["catalog-start", "catalog-ready", "paid-runtime", "paid-runtime", "paid-runtime"]); +}); + +test("local Dreamina terminal projection keeps the latest receipt, observation, and durable outputs", async () => { + const updates: GenerationTask[] = []; + const timestamp = "2026-08-18T00:00:00.000Z"; + await runBackendGenerationTask( + { + mode: "video", + prompt: "late result", + config: { ...defaultConfig, model: "local:dreamina-cli:seedance2.0mini", videoSeconds: "4", vquality: "720" }, + onTaskUpdate: (task) => updates.push(task), + }, + { + createTask: async () => { + throw new Error("must not create backend task"); + }, + waitTask: async () => { + throw new Error("must not wait backend task"); + }, + runLocal: async (input, _signal, onTaskUpdate) => { + onTaskUpdate?.({ + id: input.idempotencyKey!, + provider: "dreamina-cli", + mode: "video", + operation: "text2video", + model: "seedance2.0mini", + status: "succeeded", + stage: "succeeded", + progress: 100, + receiptRecorded: true, + officialStatus: "completed", + providerObservation: { source: "query_result", status: "completed", observedAt: timestamp }, + outputs: [{ outputIndex: 0, mediaType: "video", materializedAssetId: "asset-late-0001" }], + result: { mode: "video", video: { dataUrl: "data:video/mp4;base64,AAAA", mimeType: "video/mp4", bytes: 3 } }, + createdAt: timestamp, + updatedAt: timestamp, + }); + return { mode: "video", video: { dataUrl: "data:video/mp4;base64,AAAA", mimeType: "video/mp4", bytes: 3 } }; + }, + createId: () => "dreamina-latest-context-0001", + now: () => timestamp, + }, + ); + + expect(updates.at(-1)).toMatchObject({ + status: "succeeded", + stage: "local_cli_succeeded", + receiptRecorded: true, + officialStatus: "completed", + outputs: [{ outputIndex: 0, mediaType: "video", materializedAssetId: "asset-late-0001" }], + }); + expect(updates.at(-1)?.resultJson).toContain('"mode":"video"'); +}); + test("an explicitly cancelled Dreamina task stays cancelled instead of being projected as failed", async () => { const updates: Array<{ status: string; stage?: string }> = []; const timestamp = "2026-08-12T00:00:00.000Z"; diff --git a/web/test/local-dreamina-model-catalog.test.ts b/web/test/local-dreamina-model-catalog.test.ts index dcc29dce9..f6ea34712 100644 --- a/web/test/local-dreamina-model-catalog.test.ts +++ b/web/test/local-dreamina-model-catalog.test.ts @@ -213,6 +213,125 @@ test("effective config projects an asynchronously arriving Dreamina catalog with expect(defaultConfig.channels.some((channel) => channel.id === "local:dreamina-cli")).toBe(false); }); +test("Dreamina catalog readiness is loading immediately and shares one in-flight bootstrap", async () => { + const module = await import("../src/stores/use-local-dreamina-model-store").catch(() => ({})); + const createStore = ( + module as { + createLocalDreaminaModelStore?: (dependencies: Record) => { + getState(): { + state: string; + models: Array<{ id: string }>; + ensureReady(signal?: AbortSignal): Promise>; + }; + }; + } + ).createLocalDreaminaModelStore; + expect(typeof createStore).toBe("function"); + if (!createStore) return; + + let requests = 0; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const store = createStore({ + getRuntimeState: () => ({ connection: "connected", modules: [{ id: "dreamina", scopes: ["dreamina:models"] }] }), + getClient: () => ({ request: async () => catalogResponse() }), + loadSnapshot: async () => { + requests += 1; + await gate; + return { + accountBinding: "a".repeat(64), + sessionEpoch: 3, + models: [ + { + provider: "dreamina-cli", + id: "seedance2.0mini", + displayName: "seedance2.0mini", + modality: "video", + operations: ["text-to-video"], + adapterSupported: true, + accountEntitlement: "unknown", + currentlyObservedAvailable: "unknown", + settings: { aliases: [], aspects: ["16:9"], maxReferenceImages: 9, minDuration: 4, maxDuration: 15, tiers: ["720p"] }, + source: "runtime-execution-contract", + }, + ], + }; + }, + }); + + const first = store.getState().ensureReady(); + const second = store.getState().ensureReady(); + expect(store.getState().state).toBe("loading"); + expect(requests).toBe(1); + release(); + const [firstModels, secondModels] = await Promise.all([first, second]); + expect(firstModels.map((item) => item.id)).toEqual(["seedance2.0mini"]); + expect(secondModels).toEqual(firstModels); + expect(store.getState()).toMatchObject({ state: "ready", cacheScope: `${"a".repeat(64)}:3` }); + expect(requests).toBe(1); +}); + +test("Dreamina catalog timeout releases a hung shared request so generation can recover without reload", async () => { + const { createLocalDreaminaModelStore } = await import("../src/stores/use-local-dreamina-model-store"); + let requests = 0; + const store = createLocalDreaminaModelStore({ + getRuntimeState: () => ({ connection: "connected", modules: [{ id: "dreamina", scopes: ["dreamina:models"] }] }), + getClient: () => ({ request: async () => catalogResponse() }), + requestTimeoutMs: 10, + loadSnapshot: async (_client, signal) => { + requests += 1; + if (requests > 1) return catalogSnapshot("seedance-recovered", 4); + if (!signal) return await new Promise>(() => undefined); + return await new Promise>((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError")), { once: true }); + }); + }, + }); + + const caller = new AbortController(); + const first = store.getState().ensureReady(caller.signal); + caller.abort(); + await expect(first).rejects.toMatchObject({ name: "AbortError" }); + await Bun.sleep(20); + + const recovered = await Promise.race([store.getState().ensureReady(), Bun.sleep(40).then(() => "stuck" as const)]); + expect(recovered).not.toBe("stuck"); + expect(recovered).toMatchObject([{ id: "seedance-recovered" }]); + expect(store.getState()).toMatchObject({ state: "ready", cacheScope: `${"a".repeat(64)}:4` }); + expect(requests).toBe(2); +}); + +test("Dreamina catalog forced resync ignores a late result from the previous Runtime scope", async () => { + const { createLocalDreaminaModelStore } = await import("../src/stores/use-local-dreamina-model-store"); + let requests = 0; + let releaseOld!: (snapshot: ReturnType) => void; + const oldRequest = new Promise>((resolve) => { + releaseOld = resolve; + }); + const store = createLocalDreaminaModelStore({ + getRuntimeState: () => ({ connection: "connected", modules: [{ id: "dreamina", scopes: ["dreamina:models"] }] }), + getClient: () => ({ request: async () => catalogResponse() }), + requestTimeoutMs: 100, + loadSnapshot: async () => { + requests += 1; + return requests === 1 ? oldRequest : catalogSnapshot("seedance-current", 5); + }, + }); + + const oldSync = store.getState().sync(); + const currentSync = store.getState().sync(); + const currentResult = await Promise.race([currentSync.then(() => "ready" as const), Bun.sleep(40).then(() => "stuck" as const)]); + expect(currentResult).toBe("ready"); + expect(requests).toBe(2); + expect(store.getState()).toMatchObject({ state: "ready", cacheScope: `${"a".repeat(64)}:5`, models: [{ id: "seedance-current" }] }); + + releaseOld(catalogSnapshot("seedance-stale", 3)); + await oldSync; + expect(store.getState()).toMatchObject({ state: "ready", cacheScope: `${"a".repeat(64)}:5`, models: [{ id: "seedance-current" }] }); +}); + test("effective config removes custom channels when administrators disable them", async () => { const { createModelChannel, effectiveConfigForCustomChannels, normalizeConfigSnapshot } = await import("../src/stores/use-config-store"); const config = normalizeConfigSnapshot({ @@ -264,3 +383,24 @@ function catalogResponse() { { status: 200, headers: { "content-type": "application/json" } }, ); } + +function catalogSnapshot(id: string, sessionEpoch: number) { + return { + accountBinding: "a".repeat(64), + sessionEpoch, + models: [ + { + provider: "dreamina-cli" as const, + id, + displayName: id, + modality: "video" as const, + operations: ["text-to-video" as const], + adapterSupported: true, + accountEntitlement: "unknown" as const, + currentlyObservedAvailable: "unknown" as const, + settings: { aliases: [], aspects: ["16:9"], maxReferenceImages: 9, minDuration: 4, maxDuration: 15, tiers: ["720p"] }, + source: "runtime-execution-contract" as const, + }, + ], + }; +}