From 92647c378978286a70a8ddad622120575e64e608 Mon Sep 17 00:00:00 2001 From: 134 Date: Thu, 6 Aug 2026 12:28:50 +0800 Subject: [PATCH] fix(claw): persist and recover background chat tasks --- app/api/canvas/clawd-chat/cancel/route.ts | 12 +- app/api/canvas/clawd-chat/route.ts | 1160 +++++++---------- app/api/canvas/clawd-chat/status/route.ts | 101 ++ components/canvas/CanvasChatPanel.tsx | 431 ++++-- lib/db/queries.ts | 24 + lib/openclaw-chat/task-state.ts | 120 ++ lib/openclaw-chat/task-store.ts | 258 ++++ playwright.helpers.config.ts | 2 + .../openclaw-durable-task-state.test.ts | 45 + 9 files changed, 1339 insertions(+), 814 deletions(-) create mode 100644 app/api/canvas/clawd-chat/status/route.ts create mode 100644 lib/openclaw-chat/task-state.ts create mode 100644 lib/openclaw-chat/task-store.ts create mode 100644 tests/routes/openclaw-durable-task-state.test.ts diff --git a/app/api/canvas/clawd-chat/cancel/route.ts b/app/api/canvas/clawd-chat/cancel/route.ts index fc9d08fda..970b4ed52 100644 --- a/app/api/canvas/clawd-chat/cancel/route.ts +++ b/app/api/canvas/clawd-chat/cancel/route.ts @@ -10,6 +10,7 @@ import { OPENCLAW_AUTH_REQUIRED_CODE, OPENCLAW_AUTH_REQUIRED_MESSAGE, } from "@/lib/openclaw-chat/errors"; +import { deletePersistedClawdTask } from "@/lib/openclaw-chat/task-store"; const cancelSchema = z.object({ id: z.string().uuid(), @@ -46,5 +47,14 @@ export async function POST(request: Request) { }) ); - return NextResponse.json({ canceled: true, active }); + const deleted = await deletePersistedClawdTask({ + chatId: parsed.data.id, + userMessageId: parsed.data.userMessageId, + }).catch(() => []); + + return NextResponse.json({ + canceled: true, + active, + deleted: deleted.length > 0, + }); } diff --git a/app/api/canvas/clawd-chat/route.ts b/app/api/canvas/clawd-chat/route.ts index b7c7a7ebe..95f9bb3ec 100644 --- a/app/api/canvas/clawd-chat/route.ts +++ b/app/api/canvas/clawd-chat/route.ts @@ -1,14 +1,19 @@ -import { NextResponse } from "next/server"; +import { after, NextResponse } from "next/server"; import { z } from "zod"; import { auth } from "@/app/(auth)/auth"; import { getChatById, - getMessagesByChatId, getSystemSetting, getUserById, saveChat, - saveMessages, } from "@/lib/db/queries"; +import { + buildClawdCancelKey, + clearClawdRequestCancellation, + isClawdRequestCanceled, + registerActiveClawdRequest, + unregisterActiveClawdRequest, +} from "@/lib/openclaw-chat/cancel-registry"; import { normalizeOpenClawChatConfig, OPENCLAW_CHAT_CONFIG_KEY, @@ -19,25 +24,30 @@ import { OPENCLAW_UPGRADE_RETRY_MESSAGE, } from "@/lib/openclaw-chat/errors"; import { - buildClawdCancelKey, - cancelClawdRequest, - clearClawdRequestCancellation, - isClawdRequestCanceled, - registerActiveClawdRequest, - unregisterActiveClawdRequest, -} from "@/lib/openclaw-chat/cancel-registry"; + normalizeOpenClawChatMediaAttachments, + OPENCLAW_CHAT_MEDIA_MAX_ATTACHMENTS, + type OpenClawChatMediaAttachment, +} from "@/lib/openclaw-chat/media-attachments"; import { buildOpenClawDisplayMessage, buildOpenClawMessageWithSpreadsheetMarkdown, buildOpenClawSpreadsheetMarkdown, } from "@/lib/openclaw-chat/spreadsheet-markdown"; import { - normalizeOpenClawChatMediaAttachments, - OPENCLAW_CHAT_MEDIA_MAX_ATTACHMENTS, - type OpenClawChatMediaAttachment, -} from "@/lib/openclaw-chat/media-attachments"; + appendPersistedClawdTaskProgress, + getPersistedClawdTask, + markClawdTaskCompleted, + markClawdTaskFailed, + markClawdTaskRunning, + persistPendingClawdTask, +} from "@/lib/openclaw-chat/task-store"; import { generateUUID } from "@/lib/utils"; +const DEFAULT_PROCESSING_HINT = + "已收到，这个任务稍微繁重一些，处理时间可能会长一点，请耐心等候。"; + +export const maxDuration = 600; + const requestSchema = z.object({ id: z.string().uuid(), message: z.string().trim().max(8000).optional().default(""), @@ -68,6 +78,40 @@ type ClawdSpreadsheetAttachment = { truncated: boolean; }; +type OpenClawBackgroundTask = { + jobId: string; + state: string; + stage: string; + phase: string; + displayText: string; + updatedAt: string; + completed: number; + total: number | null; + progressPercent: number | null; + pollEndpoint: string; + pollAfterMs: number; + maxWaitSeconds: number; +}; + +type ExecuteClawdTaskParams = { + assistantMessageId: string; + cancelKey: string; + chatId: string; + config: { baseUrl: string; token: string; timeoutMs: number }; + conversationId: string; + departmentName: string; + metadata: Record | undefined; + mediaAttachments: OpenClawChatMediaAttachment[]; + skillTagId: string; + skillTagName: string; + spreadsheetFiles: unknown[]; + spreadsheetMarkdownTruncated: boolean; + upstreamMessage: string; + userId: string; + userMessageId: string; + userName: string; +}; + function buildChatTitle(message: string) { const normalized = message.replace(/\s+/g, " ").trim(); if (!normalized) return "Claw 对话"; @@ -83,17 +127,15 @@ function getOpenClawUserName(user: { name?: string | null; email?: string | null; }) { - const name = String(user.name || "").trim(); - if (name) return name; - - const email = String(user.email || "").trim(); - if (email) return email; - - return user.id; + return String(user.name || user.email || user.id).trim(); } -function getOpenClawDepartmentName(department?: string | null) { - return String(department || "").trim(); +function getMetadataString( + metadata: Record | undefined, + key: string +) { + const value = metadata?.[key]; + return typeof value === "string" ? value.trim() : ""; } function isRequestFile(value: FormDataEntryValue): value is File { @@ -116,41 +158,30 @@ async function parseClawdChatRequest( const attachments = normalizeOpenClawChatMediaAttachments( parsed.data.attachments ); - if (attachments.length !== parsed.data.attachments.length) { - return null; - } + if (attachments.length !== parsed.data.attachments.length) return null; return { ...parsed.data, attachments, tableFiles: [] }; } const formData = await request.formData().catch(() => null); if (!formData) return null; - const metadataValue = formData.get("metadata"); - const attachmentsValue = formData.get("attachments"); let metadata: Record | undefined; let attachments: unknown[] = []; - if (typeof metadataValue === "string" && metadataValue.trim()) { - try { + const metadataValue = formData.get("metadata"); + const attachmentsValue = formData.get("attachments"); + try { + if (typeof metadataValue === "string" && metadataValue.trim()) { const parsedMetadata = JSON.parse(metadataValue); - if (parsedMetadata && typeof parsedMetadata === "object") { - metadata = parsedMetadata as Record; - } else { - return null; - } - } catch { - return null; + if (!parsedMetadata || typeof parsedMetadata !== "object") return null; + metadata = parsedMetadata as Record; } - } - if (typeof attachmentsValue === "string" && attachmentsValue.trim()) { - try { + if (typeof attachmentsValue === "string" && attachmentsValue.trim()) { const parsedAttachments = JSON.parse(attachmentsValue); - if (!Array.isArray(parsedAttachments)) { - return null; - } + if (!Array.isArray(parsedAttachments)) return null; attachments = parsedAttachments; - } catch { - return null; } + } catch { + return null; } const parsed = requestSchema.safeParse({ @@ -169,88 +200,37 @@ async function parseClawdChatRequest( return null; } - const tableFiles = formData - .getAll("tableFiles") - .filter((entry): entry is File => isRequestFile(entry) && entry.size > 0); - return { ...parsed.data, attachments: normalizedAttachments, - tableFiles, + tableFiles: formData + .getAll("tableFiles") + .filter((entry): entry is File => isRequestFile(entry) && entry.size > 0), }; } async function readOpenClawConfig() { - const setting = await getSystemSetting({ key: OPENCLAW_CHAT_CONFIG_KEY }).catch( - () => null - ); + const setting = await getSystemSetting({ + key: OPENCLAW_CHAT_CONFIG_KEY, + }).catch(() => null); return normalizeOpenClawChatConfig(setting?.value); } -function getOpenClawErrorCode(payload: unknown) { - if (!payload || typeof payload !== "object") return ""; - const raw = payload as Record; - const nestedError = - raw.error && typeof raw.error === "object" - ? (raw.error as Record) - : null; - const code = nestedError?.code || raw.code; - return typeof code === "string" ? code.trim() : ""; -} - -function getMetadataString( - metadata: Record | undefined, - key: string -) { - const value = metadata?.[key]; - return typeof value === "string" ? value.trim() : ""; -} - -function createCanceledResponse() { - return NextResponse.json({ canceled: true }, { status: 499 }); -} - -function encodeSseEvent(event: string, payload: unknown) { - return `event: ${event}\ndata: ${JSON.stringify(payload)}\n\n`; -} - function parseSseBlock(block: string) { let event = "message"; const dataLines: string[] = []; for (const line of block.split(/\r?\n/)) { - if (line.startsWith("event:")) { - event = line.slice(6).trim() || "message"; - } else if (line.startsWith("data:")) { - dataLines.push(line.slice(5).trimStart()); - } + if (line.startsWith("event:")) event = line.slice(6).trim() || "message"; + if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart()); } if (dataLines.length === 0) return null; try { - return { - event, - data: JSON.parse(dataLines.join("\n")) as unknown, - }; + return { event, data: JSON.parse(dataLines.join("\n")) as unknown }; } catch { return null; } } -type OpenClawBackgroundTask = { - jobId: string; - state: string; - stage: string; - phase: string; - displayText: string; - updatedAt: string; - completed: number; - total: number | null; - progressPercent: number | null; - pollEndpoint: string; - pollAfterMs: number; - maxWaitSeconds: number; - autoNotify: boolean; -}; - function parseOpenClawBackgroundTask( payload: unknown ): OpenClawBackgroundTask | null { @@ -268,21 +248,12 @@ function parseOpenClawBackgroundTask( const jobId = String(rawTask.job_id || "").trim(); const state = String(rawTask.state || "").trim(); - const pollEndpoint = String(rawTask.poll_endpoint || "").trim(); if (!jobId || !state) return null; - const completed = Math.max(0, Number(rawTask.completed) || 0); const parsedTotal = Number(rawTask.total); const total = Number.isFinite(parsedTotal) && parsedTotal > 0 ? parsedTotal : null; - const parsedProgressPercent = Number(rawTask.progress_percent); - const progressPercent = - rawTask.progress_percent != null && - Number.isFinite(parsedProgressPercent) - ? Math.max(0, Math.min(100, parsedProgressPercent)) - : total - ? Math.round((completed * 100) / total) - : null; + const parsedPercent = Number(rawTask.progress_percent); return { jobId, @@ -295,8 +266,13 @@ function parseOpenClawBackgroundTask( updatedAt: String(rawTask.updated_at || "").trim(), completed, total, - progressPercent, - pollEndpoint, + progressPercent: + rawTask.progress_percent != null && Number.isFinite(parsedPercent) + ? Math.max(0, Math.min(100, parsedPercent)) + : total + ? Math.round((completed * 100) / total) + : null, + pollEndpoint: String(rawTask.poll_endpoint || "").trim(), pollAfterMs: Math.max( 800, Math.min(10_000, Number(rawTask.poll_after_ms) || 1500) @@ -305,11 +281,14 @@ function parseOpenClawBackgroundTask( 1, Math.min(25, Number(rawTask.max_wait_seconds) || 20) ), - autoNotify: rawTask.auto_notify === true, }; } function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) { + const message = + task.displayText || + task.phase || + (task.state === "processing" ? "任务正在处理" : "任务状态已更新"); return { step_code: task.stage || "background_task", step_state: @@ -318,14 +297,8 @@ function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) { : task.state === "processing" ? "started" : "completed", - message: - task.displayText || - task.phase || - (task.state === "processing" ? "任务正在处理" : "任务状态已更新"), - display: - task.displayText || - task.phase || - (task.state === "processing" ? "任务正在处理" : "任务状态已更新"), + message, + display: message, completed: task.completed, ...(task.total != null ? { total: task.total } : {}), ...(task.progressPercent != null @@ -358,315 +331,306 @@ function buildOpenClawTaskStatusUrl(params: { return url; } -async function persistClawdResponse({ - assistantMessageId, - attachments, - chatId, - conversationId, - displayMessage, - skillTagId, - skillTagName, - spreadsheetAttachments, - upstreamPayload, - userMessageId, -}: { - assistantMessageId: string; - attachments: OpenClawChatMediaAttachment[]; - chatId: string; - conversationId: string; - displayMessage: string; - skillTagId: string; - skillTagName: string; - spreadsheetAttachments: ClawdSpreadsheetAttachment[]; - upstreamPayload: Record; - userMessageId: string; -}) { - const answer = String(upstreamPayload.answer || "").trim(); - if (!answer) { - throw new Error("OpenClaw returned an empty answer"); +function getAnswer(payload: unknown) { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) { + return ""; } - - const existingMessages = await getMessagesByChatId({ id: chatId }).catch( - () => [] - ); - const latestMessageTimestamp = existingMessages.reduce((latest, item) => { - const timestamp = - item.createdAt instanceof Date - ? item.createdAt.getTime() - : Date.parse(String(item.createdAt || "")); - return Number.isFinite(timestamp) ? Math.max(latest, timestamp) : latest; - }, 0); - const userCreatedAt = new Date( - Math.max(Date.now(), latestMessageTimestamp + 1) - ); - const assistantCreatedAt = new Date(userCreatedAt.getTime() + 1); - - const userMessage = { - id: userMessageId, - chatId, - role: "user", - parts: [{ type: "text", text: displayMessage }], - attachments: attachments.map((attachment) => ({ - name: - attachment.name || - (attachment.type === "video" ? "视频附件" : "图片附件"), - url: attachment.url, - contentType: - attachment.content_type || - (attachment.type === "video" ? "video/mp4" : "image/jpeg"), - })), - createdAt: userCreatedAt, - videoStorageStatus: null, - temporaryVideoUrl: null, - permanentVideoUrl: null, - archivedAt: null, - videoMetadata: { - composerMode: "clawd", - source: "openclaw", - imageAttachmentCount: attachments.filter( - (attachment) => attachment.type === "image" - ).length, - videoAttachmentCount: attachments.filter( - (attachment) => attachment.type === "video" - ).length, - spreadsheetAttachmentCount: spreadsheetAttachments.length, - spreadsheetAttachments, - skillTagId: skillTagId || null, - skillTagName: skillTagName || null, - }, - }; - const assistantMessage = { - id: assistantMessageId, - chatId, - role: "assistant", - parts: [{ type: "text", text: answer }], - attachments: [], - createdAt: assistantCreatedAt, - videoStorageStatus: null, - temporaryVideoUrl: null, - permanentVideoUrl: null, - archivedAt: null, - videoMetadata: { - composerMode: "clawd", - source: "openclaw", - requestId: upstreamPayload.request_id || null, - }, - }; - - await saveMessages({ messages: [userMessage, assistantMessage] }); - - return { - conversationId, - requestId: upstreamPayload.request_id || null, - messages: [ - { - id: userMessage.id, - role: "user", - parts: userMessage.parts, - createdAt: userMessage.createdAt.toISOString(), - metadata: { - createdAt: userMessage.createdAt.toISOString(), - composerMode: "clawd", - spreadsheetAttachments, - }, - attachments: userMessage.attachments, - videoMetadata: userMessage.videoMetadata, - }, - { - id: assistantMessage.id, - role: "assistant", - parts: assistantMessage.parts, - createdAt: assistantMessage.createdAt.toISOString(), - metadata: { - createdAt: assistantMessage.createdAt.toISOString(), - composerMode: "clawd", - }, - videoMetadata: assistantMessage.videoMetadata, - }, - ], - }; + return String((payload as Record).answer || "").trim(); } -function streamOpenClawBackgroundTask(params: { - request: Request; - cancelKey: string; - config: { baseUrl: string; token: string; timeoutMs: number }; +function getRequestId(payload: unknown) { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) { + return null; + } + const value = (payload as Record).request_id; + return value ? String(value) : null; +} + +async function readOpenClawSse(params: { + response: Response; + chatId: string; + assistantMessageId: string; +}) { + const reader = params.response.body?.getReader(); + if (!reader) throw new Error("OpenClaw stream is unavailable"); + const decoder = new TextDecoder(); + let buffer = ""; + let finalPayload: Record | null = null; + + const processBlock = async (block: string) => { + const event = parseSseBlock(block); + if (!event) return; + if (event.event === "progress") { + if (event.data && typeof event.data === "object") { + await appendPersistedClawdTaskProgress({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + progress: event.data as Record, + }); + } + return; + } + if ( + event.event === "final" && + event.data && + typeof event.data === "object" && + !Array.isArray(event.data) + ) { + finalPayload = event.data as Record; + } + if (event.event === "error") { + throw new Error("OpenClaw stream returned an error"); + } + }; + + while (true) { + const { done, value } = await reader.read(); + buffer += decoder.decode(value || new Uint8Array(), { stream: !done }); + const blocks = buffer.split(/\r?\n\r?\n/); + buffer = blocks.pop() || ""; + for (const block of blocks) { + await processBlock(block); + } + if (done) break; + } + if (buffer.trim()) await processBlock(buffer); + + if (!finalPayload) throw new Error("OpenClaw stream ended without a result"); + return finalPayload; +} + +async function pollOpenClawBackgroundTask(params: { + config: ExecuteClawdTaskParams["config"]; initialPayload: Record; initialTask: OpenClawBackgroundTask; - assistantMessageId: string; - attachments: OpenClawChatMediaAttachment[]; + controller: AbortController; chatId: string; - conversationId: string; - displayMessage: string; - skillTagId: string; - skillTagName: string; - spreadsheetAttachments: ClawdSpreadsheetAttachment[]; + assistantMessageId: string; userId: string; - userMessageId: string; + conversationId: string; }) { - const encoder = new TextEncoder(); - const stream = new ReadableStream({ - async start(streamController) { - const controller = new AbortController(); - const handleRequestAbort = () => controller.abort(); - params.request.signal.addEventListener("abort", handleRequestAbort, { - once: true, - }); - registerActiveClawdRequest( - params.cancelKey, - controller, - Math.max(params.config.timeoutMs, 60 * 60 * 1000) - ); + let task = params.initialTask; + let finalPayload = params.initialPayload; + let failureCount = 0; - let task = params.initialTask; - let finalPayload = params.initialPayload; - let lastProgressKey = ""; - let consecutiveFailures = 0; + await appendPersistedClawdTaskProgress({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + progress: buildOpenClawTaskProgress(task), + }); - const emitProgress = (nextTask: OpenClawBackgroundTask) => { - const progress = buildOpenClawTaskProgress(nextTask); - const key = [ - nextTask.state, - nextTask.stage, - nextTask.updatedAt, - nextTask.completed, - nextTask.total ?? "", - progress.message, - ].join("|"); - if (key === lastProgressKey) return; - lastProgressKey = key; - streamController.enqueue( - encoder.encode(encodeSseEvent("progress", progress)) - ); - }; - - try { - emitProgress(task); - while (task.state === "processing" && task.autoNotify) { - if ( - controller.signal.aborted || - isClawdRequestCanceled(params.cancelKey) - ) { - throw new DOMException("Claw request canceled", "AbortError"); - } - - const statusUrl = buildOpenClawTaskStatusUrl({ - baseUrl: params.config.baseUrl, - task, - userId: params.userId, - conversationId: params.conversationId, - }); - let statusResponse: Response; - try { - statusResponse = await fetch(statusUrl, { - headers: { - Authorization: `Bearer ${params.config.token}`, - Accept: "application/json", - }, - cache: "no-store", - signal: controller.signal, - }); - } catch (error) { - if (controller.signal.aborted) throw error; - consecutiveFailures += 1; - if (consecutiveFailures >= 5) throw error; - streamController.enqueue( - encoder.encode( - encodeSseEvent("progress", { - step_code: task.stage || "background_task", - step_state: "waiting", - message: "任务仍在运行，正在重新连接进度服务", - display: "任务仍在运行，正在重新连接进度服务", - completed: task.completed, - ...(task.total != null ? { total: task.total } : {}), - ...(task.progressPercent != null - ? { progress_percent: task.progressPercent } - : {}), - }) - ) - ); - await new Promise((resolve) => - setTimeout(resolve, task.pollAfterMs) - ); - continue; - } - - if (!statusResponse.ok) { - throw new Error(`OpenClaw task status failed: ${statusResponse.status}`); - } - const statusPayload = (await statusResponse.json().catch(() => null)) as - | Record - | null; - const nextTask = parseOpenClawBackgroundTask(statusPayload); - if (!statusPayload || !nextTask) { - throw new Error("OpenClaw task status payload is invalid"); - } - - consecutiveFailures = 0; - finalPayload = { - ...statusPayload, - request_id: - statusPayload.request_id || params.initialPayload.request_id || null, - }; - task = nextTask; - emitProgress(task); - } - - const persisted = await persistClawdResponse({ - assistantMessageId: params.assistantMessageId, - attachments: params.attachments, - chatId: params.chatId, + while (task.state === "processing") { + if (!task.pollEndpoint) { + throw new Error("OpenClaw background task has no status endpoint"); + } + try { + const response = await fetch( + buildOpenClawTaskStatusUrl({ + baseUrl: params.config.baseUrl, + task, + userId: params.userId, conversationId: params.conversationId, - displayMessage: params.displayMessage, - skillTagId: params.skillTagId, - skillTagName: params.skillTagName, - spreadsheetAttachments: params.spreadsheetAttachments, - upstreamPayload: finalPayload, - userMessageId: params.userMessageId, - }); - clearClawdRequestCancellation(params.cancelKey); - streamController.enqueue( - encoder.encode(encodeSseEvent("final", persisted)) - ); - } catch (error) { - const wasCanceled = - controller.signal.aborted || - isClawdRequestCanceled(params.cancelKey) || - (error instanceof Error && error.name === "AbortError"); - streamController.enqueue( - encoder.encode( - encodeSseEvent( - wasCanceled ? "final" : "error", - wasCanceled - ? { canceled: true } - : { error: OPENCLAW_UPGRADE_RETRY_MESSAGE } - ) - ) - ); - } finally { - params.request.signal.removeEventListener("abort", handleRequestAbort); - unregisterActiveClawdRequest(params.cancelKey); - try { - streamController.close(); - } catch { - // The browser may already have closed the stream. + }), + { + headers: { + Authorization: `Bearer ${params.config.token}`, + Accept: "application/json", + }, + cache: "no-store", + signal: params.controller.signal, } + ); + if (!response.ok) { + throw new Error(`OpenClaw task status failed: ${response.status}`); } - }, - }); + const payload = (await response.json().catch(() => null)) as + | Record + | null; + const nextTask = parseOpenClawBackgroundTask(payload); + if (!payload || !nextTask) { + throw new Error("OpenClaw task status payload is invalid"); + } + failureCount = 0; + finalPayload = { + ...payload, + request_id: + payload.request_id || params.initialPayload.request_id || null, + }; + task = nextTask; + await appendPersistedClawdTaskProgress({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + progress: buildOpenClawTaskProgress(task), + }); + } catch (error) { + if (params.controller.signal.aborted) throw error; + failureCount += 1; + if (failureCount >= 5) throw error; + await appendPersistedClawdTaskProgress({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + progress: { + step_code: task.stage || "background_task", + step_state: "waiting", + message: "任务仍在运行，正在重新连接进度服务", + display: "任务仍在运行，正在重新连接进度服务", + }, + }); + await new Promise((resolve) => setTimeout(resolve, task.pollAfterMs)); + } + } + return finalPayload; +} - return new Response(stream, { - headers: { - "Cache-Control": "no-cache, no-store", - "Content-Type": "text/event-stream; charset=utf-8", - "X-Accel-Buffering": "no", +async function executeClawdTask(params: ExecuteClawdTaskParams) { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), params.config.timeoutMs); + registerActiveClawdRequest( + params.cancelKey, + controller, + Math.max(params.config.timeoutMs, 10 * 60 * 1000) + ); + + try { + if (isClawdRequestCanceled(params.cancelKey)) return; + await markClawdTaskRunning({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + }); + + const response = await fetch(`${params.config.baseUrl}/api/chat`, { + method: "POST", + headers: { + Authorization: `Bearer ${params.config.token}`, + "Content-Type": "application/json", + Accept: "text/event-stream", + }, + body: JSON.stringify({ + conversation_id: params.conversationId, + user_id: params.userId, + user_name: params.userName, + username: params.userName, + department: params.departmentName, + department_name: params.departmentName, + user_department: params.departmentName, + skill_tag_id: params.skillTagId || undefined, + skill_tag_name: params.skillTagName || undefined, + message: params.upstreamMessage, + attachments: + params.mediaAttachments.length > 0 + ? params.mediaAttachments + : undefined, + metadata: { + source: "ai_chatbot_canvas", + page: "/canvas", + tenant: "default", + chat_id: params.chatId, + user_id: params.userId, + user_name: params.userName, + username: params.userName, + department: params.departmentName, + department_name: params.departmentName, + user_department: params.departmentName, + skill_tag_id: params.skillTagId || undefined, + skill_tag_name: params.skillTagName || undefined, + media_attachments: + params.mediaAttachments.length > 0 + ? params.mediaAttachments + : undefined, + spreadsheet_files: + params.spreadsheetFiles.length > 0 + ? params.spreadsheetFiles + : undefined, + spreadsheet_markdown_truncated: + params.spreadsheetMarkdownTruncated || undefined, + ...params.metadata, + }, + stream_progress: true, + }), + signal: controller.signal, + cache: "no-store", + }); + + if (!response.ok) { + throw new Error(`OpenClaw request failed: ${response.status}`); + } + + const contentType = response.headers.get("content-type") || ""; + let payload = contentType.includes("text/event-stream") + ? await readOpenClawSse({ + response, + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + }) + : ((await response.json().catch(() => null)) as Record< + string, + unknown + > | null); + if (!payload) throw new Error("OpenClaw returned an invalid response"); + + const backgroundTask = parseOpenClawBackgroundTask(payload); + if (backgroundTask?.state === "processing") { + payload = await pollOpenClawBackgroundTask({ + config: params.config, + initialPayload: payload, + initialTask: backgroundTask, + controller, + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + userId: params.userId, + conversationId: params.conversationId, + }); + } + + if (isClawdRequestCanceled(params.cancelKey)) return; + const answer = getAnswer(payload); + if (!answer) throw new Error("OpenClaw returned an empty answer"); + await markClawdTaskCompleted({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + answer, + requestId: getRequestId(payload), + }); + clearClawdRequestCancellation(params.cancelKey); + } catch (error) { + if (isClawdRequestCanceled(params.cancelKey)) return; + console.error("[ClawdChat API] Durable task failed:", error); + await markClawdTaskFailed({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + error: OPENCLAW_UPGRADE_RETRY_MESSAGE, + }).catch(() => null); + } finally { + clearTimeout(timeout); + unregisterActiveClawdRequest(params.cancelKey); + } +} + +function buildTaskResponse(params: { + chatId: string; + userMessageId: string; + assistantMessageId: string; + status: string; + processingHint: string; +}) { + const statusUrl = `/api/canvas/clawd-chat/status?id=${encodeURIComponent( + params.chatId + )}&userMessageId=${encodeURIComponent(params.userMessageId)}`; + return { + accepted: true, + task: { + chatId: params.chatId, + userMessageId: params.userMessageId, + assistantMessageId: params.assistantMessageId, + status: params.status, + processingHint: params.processingHint, + statusUrl, }, - }); + }; } export async function POST(request: Request) { const session = await auth().catch(() => null); - if (!session?.user?.id) { return NextResponse.json( { @@ -693,13 +657,7 @@ export async function POST(request: Request) { } const config = await readOpenClawConfig(); - if (!config.enabled) { - return NextResponse.json( - { error: OPENCLAW_UPGRADE_RETRY_MESSAGE }, - { status: 503 } - ); - } - if (!config.baseUrl || !config.token) { + if (!config.enabled || !config.baseUrl || !config.token) { return NextResponse.json( { error: OPENCLAW_UPGRADE_RETRY_MESSAGE }, { status: 503 } @@ -712,13 +670,22 @@ export async function POST(request: Request) { } const userMessageId = parsed.userMessageId || generateUUID(); - const cancelKey = buildClawdCancelKey({ - userId: session.user.id, + const existingTask = await getPersistedClawdTask({ chatId: parsed.id, userMessageId, - }); - if (isClawdRequestCanceled(cancelKey)) { - return createCanceledResponse(); + }).catch(() => null); + if (existingTask) { + return NextResponse.json( + buildTaskResponse({ + chatId: parsed.id, + userMessageId, + assistantMessageId: existingTask.metadata.assistantMessageId, + status: existingTask.metadata.status, + processingHint: + existingTask.metadata.processingHint || DEFAULT_PROCESSING_HINT, + }), + { status: 202, headers: { "Retry-After": "2" } } + ); } let spreadsheetMarkdownResult: Awaited< @@ -766,10 +733,6 @@ export async function POST(request: Request) { }; }); - if (isClawdRequestCanceled(cancelKey)) { - return createCanceledResponse(); - } - if (!existingChat) { await saveChat({ id: parsed.id, @@ -780,273 +743,82 @@ export async function POST(request: Request) { } const assistantMessageId = generateUUID(); - const conversationId = createConversationId(session.user.id, parsed.id); - const userProfile = await getUserById({ id: session.user.id }).catch(() => null); - const openClawUserName = getOpenClawUserName({ + const skillTagId = getMetadataString(parsed.metadata, "skillTagId"); + const skillTagName = getMetadataString(parsed.metadata, "skillTagName"); + const processingHint = + getMetadataString(parsed.metadata, "processingHint") || + DEFAULT_PROCESSING_HINT; + const messageAttachments = parsed.attachments.map((attachment) => ({ + name: + attachment.name || + (attachment.type === "video" ? "视频附件" : "图片附件"), + url: attachment.url, + contentType: + attachment.content_type || + (attachment.type === "video" ? "video/mp4" : "image/jpeg"), + })); + + await persistPendingClawdTask({ + chatId: parsed.id, + userMessageId, + assistantMessageId, + displayMessage, + attachments: messageAttachments, + spreadsheetAttachments, + skillTagId, + skillTagName, + processingHint, + }); + + const userProfile = await getUserById({ id: session.user.id }).catch( + () => null + ); + const userName = getOpenClawUserName({ id: session.user.id, name: userProfile?.name ?? session.user.name, email: userProfile?.email ?? session.user.email, }); - const openClawDepartmentName = getOpenClawDepartmentName(userProfile?.department); - const skillTagId = getMetadataString(parsed.metadata, "skillTagId"); - const skillTagName = getMetadataString(parsed.metadata, "skillTagName"); - const endpoint = `${config.baseUrl}/api/chat`; - - let upstreamPayload: unknown = null; - let upstreamStatus = 502; - - if (isClawdRequestCanceled(cancelKey)) { - return createCanceledResponse(); - } - - try { - const controller = new AbortController(); - registerActiveClawdRequest(cancelKey, controller, config.timeoutMs); - const timeout = setTimeout(() => controller.abort(), config.timeoutMs); - const handleRequestAbort = () => { - cancelClawdRequest(cancelKey); - }; - request.signal.addEventListener("abort", handleRequestAbort, { once: true }); - const response = await fetch(endpoint, { - method: "POST", - headers: { - Authorization: `Bearer ${config.token}`, - "Content-Type": "application/json", - Accept: "text/event-stream", - }, - body: JSON.stringify({ - conversation_id: conversationId, - user_id: session.user.id, - user_name: openClawUserName, - username: openClawUserName, - department: openClawDepartmentName, - department_name: openClawDepartmentName, - user_department: openClawDepartmentName, - skill_tag_id: skillTagId || undefined, - skill_tag_name: skillTagName || undefined, - message: upstreamMessage, - attachments: - parsed.attachments.length > 0 ? parsed.attachments : undefined, - metadata: { - source: "ai_chatbot_canvas", - page: "/canvas", - tenant: "default", - chat_id: parsed.id, - user_id: session.user.id, - user_name: openClawUserName, - username: openClawUserName, - department: openClawDepartmentName, - department_name: openClawDepartmentName, - user_department: openClawDepartmentName, - skill_tag_id: skillTagId || undefined, - skill_tag_name: skillTagName || undefined, - media_attachments: - parsed.attachments.length > 0 ? parsed.attachments : undefined, - spreadsheet_files: spreadsheetMarkdownResult?.files || undefined, - spreadsheet_markdown_truncated: - spreadsheetMarkdownResult?.truncated || undefined, - ...parsed.metadata, - }, - stream_progress: true, - }), - signal: controller.signal, - }).finally(() => { - clearTimeout(timeout); - request.signal.removeEventListener("abort", handleRequestAbort); - unregisterActiveClawdRequest(cancelKey); - }); - - upstreamStatus = response.status; - const upstreamContentType = response.headers.get("content-type") || ""; - if ( - response.ok && - upstreamContentType.includes("text/event-stream") && - response.body - ) { - const encoder = new TextEncoder(); - const decoder = new TextDecoder(); - const stream = new ReadableStream({ - async start(streamController) { - const reader = response.body!.getReader(); - let buffer = ""; - let finalReceived = false; - try { - while (true) { - const { done, value } = await reader.read(); - buffer += decoder.decode(value || new Uint8Array(), { - stream: !done, - }); - const blocks = buffer.split(/\r?\n\r?\n/); - buffer = blocks.pop() || ""; - for (const block of blocks) { - const parsedEvent = parseSseBlock(block); - if (!parsedEvent) continue; - if (parsedEvent.event === "progress") { - streamController.enqueue( - encoder.encode(encodeSseEvent("progress", parsedEvent.data)) - ); - continue; - } - if (parsedEvent.event !== "final") continue; - const finalPayload = - parsedEvent.data && - typeof parsedEvent.data === "object" && - !Array.isArray(parsedEvent.data) - ? (parsedEvent.data as Record) - : {}; - const persisted = await persistClawdResponse({ - assistantMessageId, - attachments: parsed.attachments, - chatId: parsed.id, - conversationId, - displayMessage, - skillTagId, - skillTagName, - spreadsheetAttachments, - upstreamPayload: finalPayload, - userMessageId, - }); - clearClawdRequestCancellation(cancelKey); - streamController.enqueue( - encoder.encode(encodeSseEvent("final", persisted)) - ); - finalReceived = true; - } - if (done) break; - } - if (!finalReceived) { - streamController.enqueue( - encoder.encode( - encodeSseEvent("error", { - error: OPENCLAW_UPGRADE_RETRY_MESSAGE, - }) - ) - ); - } - } catch (error) { - console.error("[ClawdChat API] OpenClaw stream failed:", error); - try { - streamController.enqueue( - encoder.encode( - encodeSseEvent("error", { - error: OPENCLAW_UPGRADE_RETRY_MESSAGE, - }) - ) - ); - } catch { - // The browser may have closed while the upstream task keeps running. - } - } finally { - try { - streamController.close(); - } catch { - // The stream may already be canceled by the browser. - } - } - }, - }); - return new Response(stream, { - headers: { - "Cache-Control": "no-cache, no-store", - "Content-Type": "text/event-stream; charset=utf-8", - "X-Accel-Buffering": "no", - }, - }); - } - upstreamPayload = await response.json().catch(() => null); - - if (!response.ok) { - const errorCode = getOpenClawErrorCode(upstreamPayload); - console.error("[ClawdChat API] OpenClaw upstream error:", { - status: response.status, - code: errorCode || undefined, - payload: upstreamPayload, - }); - return NextResponse.json( - { - error: OPENCLAW_UPGRADE_RETRY_MESSAGE, - code: errorCode || undefined, - }, - { status: response.status } - ); - } - - const backgroundTask = parseOpenClawBackgroundTask(upstreamPayload); - if ( - backgroundTask?.state === "processing" && - backgroundTask.autoNotify && - backgroundTask.pollEndpoint - ) { - return streamOpenClawBackgroundTask({ - request, - cancelKey, - config, - initialPayload: upstreamPayload as Record, - initialTask: backgroundTask, - assistantMessageId, - attachments: parsed.attachments, - chatId: parsed.id, - conversationId, - displayMessage, - skillTagId, - skillTagName, - spreadsheetAttachments, - userId: session.user.id, - userMessageId, - }); - } - } catch (error) { - unregisterActiveClawdRequest(cancelKey); - const errorName = - error && typeof error === "object" && "name" in error - ? String((error as { name?: unknown }).name || "") - : ""; - const isAbortError = errorName === "AbortError"; - if (isClawdRequestCanceled(cancelKey)) { - console.info("[ClawdChat API] OpenClaw request canceled:", { - chatId: parsed.id, - userMessageId, - }); - return createCanceledResponse(); - } - console.error("[ClawdChat API] OpenClaw request failed:", error); - return NextResponse.json( - { - error: OPENCLAW_UPGRADE_RETRY_MESSAGE, - }, - { status: isAbortError ? 504 : upstreamStatus } - ); - } - - const answer = - upstreamPayload && typeof upstreamPayload === "object" - ? String((upstreamPayload as Record).answer || "").trim() - : ""; - - if (!answer) { - console.error("[ClawdChat API] OpenClaw empty answer:", upstreamPayload); - return NextResponse.json( - { error: OPENCLAW_UPGRADE_RETRY_MESSAGE }, - { status: 502 } - ); - } - - if (isClawdRequestCanceled(cancelKey)) { - return createCanceledResponse(); - } - - const persistedResponse = await persistClawdResponse({ - assistantMessageId, - attachments: parsed.attachments, + const cancelKey = buildClawdCancelKey({ + userId: session.user.id, chatId: parsed.id, - conversationId, - displayMessage, - skillTagId, - skillTagName, - spreadsheetAttachments, - upstreamPayload: upstreamPayload as Record, userMessageId, }); - clearClawdRequestCancellation(cancelKey); - return NextResponse.json(persistedResponse); + + after(async () => { + await executeClawdTask({ + assistantMessageId, + cancelKey, + chatId: parsed.id, + config, + conversationId: createConversationId(session.user.id, parsed.id), + departmentName: String(userProfile?.department || "").trim(), + metadata: parsed.metadata, + mediaAttachments: parsed.attachments, + skillTagId, + skillTagName, + spreadsheetFiles: spreadsheetMarkdownResult?.files || [], + spreadsheetMarkdownTruncated: + spreadsheetMarkdownResult?.truncated || false, + upstreamMessage, + userId: session.user.id, + userMessageId, + userName, + }); + }); + + const responseBody = buildTaskResponse({ + chatId: parsed.id, + userMessageId, + assistantMessageId, + status: "queued", + processingHint, + }); + return NextResponse.json(responseBody, { + status: 202, + headers: { + "Cache-Control": "no-store, max-age=0", + Location: responseBody.task.statusUrl, + "Retry-After": "2", + }, + }); } diff --git a/app/api/canvas/clawd-chat/status/route.ts b/app/api/canvas/clawd-chat/status/route.ts new file mode 100644 index 000000000..81be3d3f1 --- /dev/null +++ b/app/api/canvas/clawd-chat/status/route.ts @@ -0,0 +1,101 @@ +import { NextResponse } from "next/server"; +import { z } from "zod"; +import { auth } from "@/app/(auth)/auth"; +import { getChatById } from "@/lib/db/queries"; +import { + OPENCLAW_AUTH_REQUIRED_CODE, + OPENCLAW_AUTH_REQUIRED_MESSAGE, +} from "@/lib/openclaw-chat/errors"; +import { getPersistedClawdTask } from "@/lib/openclaw-chat/task-store"; + +const statusSchema = z.object({ + id: z.string().uuid(), + userMessageId: z.string().uuid().optional(), +}); + +export const dynamic = "force-dynamic"; +export const revalidate = 0; + +export async function GET(request: Request) { + const session = await auth().catch(() => null); + if (!session?.user?.id) { + return NextResponse.json( + { + error: OPENCLAW_AUTH_REQUIRED_MESSAGE, + code: OPENCLAW_AUTH_REQUIRED_CODE, + }, + { status: 401 } + ); + } + + const url = new URL(request.url); + const parsed = statusSchema.safeParse({ + id: url.searchParams.get("id"), + userMessageId: url.searchParams.get("userMessageId") || undefined, + }); + if (!parsed.success) { + return NextResponse.json({ error: "请求参数不正确" }, { status: 400 }); + } + + const chat = await getChatById({ id: parsed.data.id }).catch(() => null); + if (!chat) { + return NextResponse.json({ task: null }, { status: 404 }); + } + if (chat.userId !== session.user.id) { + return NextResponse.json({ error: "Forbidden" }, { status: 403 }); + } + + const task = await getPersistedClawdTask({ + chatId: parsed.data.id, + userMessageId: parsed.data.userMessageId, + }); + if (!task) { + return NextResponse.json( + { task: null }, + { headers: { "Cache-Control": "no-store, max-age=0" } } + ); + } + + const userTextPart = Array.isArray(task.userMessage?.parts) + ? task.userMessage.parts.find( + (part) => + part && + typeof part === "object" && + !Array.isArray(part) && + (part as { type?: unknown }).type === "text" + ) + : null; + const userText = + userTextPart && typeof userTextPart === "object" + ? String((userTextPart as { text?: unknown }).text || "") + : ""; + + return NextResponse.json( + { + task: { + chatId: parsed.data.id, + userMessageId: task.metadata.userMessageId, + assistantMessageId: task.metadata.assistantMessageId, + status: task.metadata.status, + processingHint: task.metadata.processingHint, + progressEvents: task.metadata.progressEvents, + skillTagId: task.metadata.skillTagId || null, + skillTagName: task.metadata.skillTagName || null, + requestId: task.metadata.requestId || null, + error: task.metadata.error || null, + createdAt: task.metadata.createdAt, + updatedAt: task.metadata.updatedAt, + userMessage: { + text: userText, + attachments: task.userMessage?.attachments || [], + }, + }, + }, + { + headers: { + "Cache-Control": "no-store, max-age=0", + "Retry-After": "2", + }, + } + ); +} diff --git a/components/canvas/CanvasChatPanel.tsx b/components/canvas/CanvasChatPanel.tsx index 1c3e95f25..fbc40e913 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -540,6 +540,137 @@ type ClawdProgressEvent = { progressPercent?: number; }; +type ClawdDurableTaskStatus = + | "queued" + | "running" + | "completed" + | "failed" + | "canceled"; + +type ClawdDurableTaskSnapshot = { + chatId: string; + userMessageId: string; + assistantMessageId: string; + status: ClawdDurableTaskStatus; + processingHint: string; + progressEvents: Record[]; + skillTagId: string | null; + skillTagName: string | null; + requestId: string | null; + error: string | null; + createdAt: number; + updatedAt: number; + userMessage: { + text: string; + attachments: Array<{ + name?: string; + url?: string; + contentType?: string; + }>; + }; +}; + +type ClawdDurableTaskResponse = { + task: ClawdDurableTaskSnapshot | null; +}; + +class ClawdTaskStatusError extends Error { + status: number; + + constructor(message: string, status: number) { + super(message); + this.name = "ClawdTaskStatusError"; + this.status = status; + } +} + +function isPendingClawdDurableTask(status: ClawdDurableTaskStatus) { + return status === "queued" || status === "running"; +} + +async function fetchClawdDurableTask(params: { + rawChatId: string; + userMessageId?: string; + signal?: AbortSignal; +}) { + const query = new URLSearchParams({ id: params.rawChatId }); + if (params.userMessageId) { + query.set("userMessageId", params.userMessageId); + } + const response = await fetch( + `/api/canvas/clawd-chat/status?${query.toString()}`, + { + cache: "no-store", + signal: params.signal, + } + ); + if (response.status === 404) return null; + const payload = (await response.json().catch(() => null)) as + | ClawdDurableTaskResponse + | null; + if (!response.ok) { + throw new ClawdTaskStatusError( + payload && "error" in payload + ? String((payload as { error?: unknown }).error || "") + : `Claw task status failed: ${response.status}`, + response.status + ); + } + return payload?.task || null; +} + +async function waitForClawdDurableTask(params: { + rawChatId: string; + userMessageId: string; + signal?: AbortSignal; + onUpdate: (task: ClawdDurableTaskSnapshot) => void; +}) { + let consecutiveFailures = 0; + while (!params.signal?.aborted) { + let task: ClawdDurableTaskSnapshot | null = null; + try { + task = await fetchClawdDurableTask({ + rawChatId: params.rawChatId, + userMessageId: params.userMessageId, + signal: params.signal, + }); + consecutiveFailures = 0; + } catch (error) { + if (params.signal?.aborted) throw error; + if ( + error instanceof ClawdTaskStatusError && + [400, 401, 403].includes(error.status) + ) { + throw error; + } + consecutiveFailures += 1; + } + if (!task && consecutiveFailures === 0) return null; + if (!task) { + await new Promise((resolve) => window.setTimeout(resolve, 1600)); + continue; + } + params.onUpdate(task); + if (!isPendingClawdDurableTask(task.status)) return task; + await new Promise((resolve, reject) => { + const cleanup = () => { + params.signal?.removeEventListener("abort", handleAbort); + }; + const timer = window.setTimeout(() => { + cleanup(); + resolve(); + }, 1600); + const handleAbort = () => { + window.clearTimeout(timer); + cleanup(); + reject(new DOMException("Claw task polling canceled", "AbortError")); + }; + params.signal?.addEventListener("abort", handleAbort, { once: true }); + }); + } + throw new DOMException("Claw task polling canceled", "AbortError"); +} + function normalizeClawdProgressEvent( progress: Record ): ClawdProgressEvent | null { @@ -690,64 +821,6 @@ function shouldShowClawdProcessingHint( ); } -async function readClawdResponse( - response: Response, - onProgress: (progress: Record) => void -) { - const contentType = response.headers.get("content-type") || ""; - if (!contentType.includes("text/event-stream") || !response.body) { - return response.json().catch(() => null); - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let finalPayload: unknown = null; - let streamError = ""; - - const consumeBlock = (block: string) => { - let event = "message"; - const dataLines: string[] = []; - for (const line of block.split(/\r?\n/)) { - if (line.startsWith("event:")) { - event = line.slice(6).trim() || "message"; - } else if (line.startsWith("data:")) { - dataLines.push(line.slice(5).trimStart()); - } - } - if (dataLines.length === 0) return; - let data: unknown; - try { - data = JSON.parse(dataLines.join("\n")); - } catch { - return; - } - if (event === "progress" && data && typeof data === "object") { - onProgress(data as Record); - } else if (event === "final") { - finalPayload = data; - } else if (event === "error") { - streamError = - data && typeof data === "object" && "error" in data - ? String((data as { error?: unknown }).error || "") - : OPENCLAW_UPGRADE_RETRY_MESSAGE; - } - }; - - while (true) { - const { done, value } = await reader.read(); - buffer += decoder.decode(value || new Uint8Array(), { stream: !done }); - const blocks = buffer.split(/\r?\n\r?\n/); - buffer = blocks.pop() || ""; - blocks.forEach(consumeBlock); - if (done) break; - } - if (buffer.trim()) consumeBlock(buffer); - if (streamError) throw new Error(streamError); - if (!finalPayload) throw new Error(OPENCLAW_UPGRADE_RETRY_MESSAGE); - return finalPayload; -} - function formatClawdTableFileSize(bytes: number) { if (!Number.isFinite(bytes) || bytes <= 0) { return "0KB"; @@ -2793,6 +2866,9 @@ export function CanvasChatPanel({ const clawdAbortControllersRef = useRef< Map >(new Map()); + const clawdRecoveryPollersRef = useRef>( + new Map() + ); const pendingClawdRequestRestoresRef = useRef< Map >(new Map()); @@ -3002,6 +3078,34 @@ export function CanvasChatPanel({ [] ); + const applyClawdDurableTaskToUi = useCallback( + (chatSessionId: string, task: ClawdDurableTaskSnapshot) => { + const normalizedProgressEvents = task.progressEvents + .map((progress) => normalizeClawdProgressEvent(progress)) + .filter((event): event is ClawdProgressEvent => Boolean(event)); + const latestProgressHint = + normalizedProgressEvents.at(-1)?.message || + task.processingHint || + CLAWD_PROCESSING_HINT; + + startClawdRequestUi( + chatSessionId, + task.userMessageId, + latestProgressHint + ); + updateClawdRequestUi( + chatSessionId, + task.userMessageId, + (currentState) => ({ + ...currentState, + hint: latestProgressHint, + progressEvents: normalizedProgressEvents, + }) + ); + }, + [startClawdRequestUi, updateClawdRequestUi] + ); + const setStreamImageGenerationStateForChat = useCallback( (chatSessionId: string, next: ImageGenerationStatusStreamPayload | null) => { setStreamImageGenerationStateByChatId((prev) => ({ @@ -5500,6 +5604,120 @@ export function CanvasChatPanel({ isDetailGenerationMode, ]); + useEffect(() => { + if (!isClawdMode || !rawChatId || !session?.user?.id) return; + if (clawdAbortControllersRef.current.has(activeChatId)) return; + + const controller = new AbortController(); + clawdRecoveryPollersRef.current.get(activeChatId)?.abort(); + clawdRecoveryPollersRef.current.set(activeChatId, controller); + + const recoverPendingTask = async () => { + try { + const pendingTask = await fetchClawdDurableTask({ + rawChatId, + signal: controller.signal, + }); + if (!pendingTask || !isPendingClawdDurableTask(pendingTask.status)) { + return; + } + + applyClawdDurableTaskToUi(activeChatId, pendingTask); + setSendGuard(activeChatId, { + inFlight: true, + signature: `recovered:${pendingTask.userMessageId}`, + timestamp: pendingTask.updatedAt || Date.now(), + }); + + if (pendingTask.skillTagId) { + markSessionClawdSkillTagPromptApplied( + activeChatId, + pendingTask.skillTagId + ); + } + + const recoveredSkillTag = pendingTask.skillTagId + ? latestCanvasSkillTagsRef.current.find( + (tag) => tag.id === pendingTask.skillTagId + ) || null + : null; + const recoveredAttachments = (pendingTask.userMessage.attachments || []) + .map((attachment) => + sanitizeChatAttachment({ + name: String(attachment.name || "附件"), + url: String(attachment.url || ""), + contentType: String( + attachment.contentType || "application/octet-stream" + ), + } as PickedObjectAttachment) + ) + .filter( + (attachment): attachment is PickedObjectAttachment => + Boolean(attachment) + ); + pendingClawdRequestRestoresRef.current.set(activeChatId, { + chatId: activeChatId, + userMessageId: pendingTask.userMessageId, + textContent: pendingTask.userMessage.text, + tableFiles: [], + attachments: recoveredAttachments, + skillTag: recoveredSkillTag, + shouldRestoreSkillTag: Boolean(recoveredSkillTag), + }); + + const terminalTask = await waitForClawdDurableTask({ + rawChatId, + userMessageId: pendingTask.userMessageId, + signal: controller.signal, + onUpdate: (task) => applyClawdDurableTaskToUi(activeChatId, task), + }); + if (!terminalTask || controller.signal.aborted) return; + + if (terminalTask.status === "completed") { + setClawdTypewriterMessageIds((previousIds) => ({ + ...previousIds, + [terminalTask.assistantMessageId]: true, + })); + } + pendingClawdRequestRestoresRef.current.delete(activeChatId); + clearClawdRequestUi(activeChatId, terminalTask.userMessageId); + releaseSendGuard(activeChatId); + if (activeChatIdRef.current === activeChatId) { + forceHistoryHydrationRef.current = rawChatId; + await mutateHistoryData(); + } + void mutateClawdHistory(); + } catch (error) { + if (controller.signal.aborted) return; + console.error("[CanvasChat] Failed to recover Claw task:", error); + } finally { + if (clawdRecoveryPollersRef.current.get(activeChatId) === controller) { + clawdRecoveryPollersRef.current.delete(activeChatId); + } + } + }; + + void recoverPendingTask(); + return () => { + controller.abort(); + if (clawdRecoveryPollersRef.current.get(activeChatId) === controller) { + clawdRecoveryPollersRef.current.delete(activeChatId); + } + }; + }, [ + activeChatId, + applyClawdDurableTaskToUi, + clearClawdRequestUi, + isClawdMode, + markSessionClawdSkillTagPromptApplied, + mutateClawdHistory, + mutateHistoryData, + rawChatId, + releaseSendGuard, + session?.user?.id, + setSendGuard, + ]); + useEffect(() => { if (!activeChatId) return; const historyMessages = historyData?.messages || []; @@ -8141,6 +8359,8 @@ export function CanvasChatPanel({ activeController.controller.abort(); clawdAbortControllersRef.current.delete(activeChatId); } + clawdRecoveryPollersRef.current.get(activeChatId)?.abort(); + clawdRecoveryPollersRef.current.delete(activeChatId); clearClawdRequestUi(activeChatId, snapshot.userMessageId); releaseSendGuard(snapshot.chatId); restorePendingClawdRequestToComposer(snapshot); @@ -11927,6 +12147,7 @@ export function CanvasChatPanel({ skillTagId: clawdSkillTagForRequest?.id || undefined, skillTagName: clawdSkillTagForRequest?.name || undefined, skillTagPromptIncluded: hasPendingClawdSkillTagPrompt || undefined, + processingHint: nextProcessingHint || CLAWD_PROCESSING_HINT, }) ); if (openClawMediaAttachments.length > 0) { @@ -11945,32 +12166,7 @@ export function CanvasChatPanel({ signal: clawdAbortController.signal, }); clawdRequestAccepted = response.ok; - const responseData = await readClawdResponse(response, (progress) => { - const display = String( - progress.display || progress.message || "" - ).trim(); - const normalizedProgress = normalizeClawdProgressEvent(progress); - if (display || normalizedProgress) { - updateClawdRequestUi( - requestChatId, - userMessageId, - (currentState) => { - const nextProgressEvents = normalizedProgress - ? currentState.progressEvents.some( - (item) => item.key === normalizedProgress.key - ) - ? currentState.progressEvents - : [...currentState.progressEvents, normalizedProgress].slice(-2000) - : currentState.progressEvents; - return { - ...currentState, - hint: display || currentState.hint, - progressEvents: nextProgressEvents, - }; - } - ); - } - }); + const responseData = await response.json().catch(() => null); if (response.status === 499 || responseData?.canceled === true) { canceledClawdMessageIdsRef.current.add(userMessageId); throw new DOMException("Claw request canceled", "AbortError"); @@ -11991,6 +12187,16 @@ export function CanvasChatPanel({ throw new Error(clawdFailureMessage); } + const acceptedTask = responseData?.task as + | Partial + | undefined; + if ( + !acceptedTask?.userMessageId || + !acceptedTask.assistantMessageId + ) { + throw new Error("Claw 任务创建失败，请稍后重试"); + } + if (clawdSkillTagPrompt && clawdSkillTagForRequest) { markSessionClawdSkillTagPromptApplied( requestChatId, @@ -11998,41 +12204,26 @@ export function CanvasChatPanel({ ); } - const returnedMessages = Array.isArray(responseData?.messages) - ? (responseData.messages as ChatMessage[]) - : []; - if ( - returnedMessages.length > 0 && - activeChatIdRef.current === requestChatId - ) { - const returnedIds = new Set(returnedMessages.map((message) => message.id)); - const returnedAssistantIds = returnedMessages - .filter((message) => message.role === "assistant") - .map((message) => message.id) - .filter(Boolean); - if (returnedAssistantIds.length > 0) { - setClawdTypewriterMessageIds((previousIds) => { - const nextIds = { ...previousIds }; - returnedAssistantIds.forEach((messageId) => { - nextIds[messageId] = true; - }); - return nextIds; - }); - } - setMessages((previousMessages) => - sortCanvasMessagesByCreatedAt([ - ...previousMessages.filter( - (message) => - message.id !== userMessageId && !returnedIds.has(message.id) - ), - ...returnedMessages, - ]) - ); + const terminalTask = await waitForClawdDurableTask({ + rawChatId: requestRawChatId, + userMessageId, + signal: clawdAbortController.signal, + onUpdate: (task) => { + applyClawdDurableTaskToUi(requestChatId, task); + }, + }); + if (!terminalTask) { + throw new Error("Claw 任务记录暂时不可用"); } - if (requestRawChatId) { - void mutate( - buildHistoryMessagesUrl(requestRawChatId, { cacheBust: true }) - ); + if (terminalTask.status === "completed") { + setClawdTypewriterMessageIds((previousIds) => ({ + ...previousIds, + [terminalTask.assistantMessageId]: true, + })); + } + if (activeChatIdRef.current === requestChatId && requestRawChatId) { + forceHistoryHydrationRef.current = requestRawChatId; + await mutateHistoryData(); } void mutateClawdHistory(); hasHydratedHistoryRef.current = true; @@ -12061,7 +12252,7 @@ export function CanvasChatPanel({ pendingClawdSubmissionRef.current = null; } if (clawdRequestAccepted) { - toast.error("Claw 连接中断，已保留已发送消息；任务可能仍在处理中，请稍后刷新历史查看"); + toast.message("Claw 任务仍在后台处理中，返回此对话会自动恢复进度"); void mutateClawdHistory(); if (requestRawChatId) { void mutate( @@ -12572,6 +12763,8 @@ export function CanvasChatPanel({ startClawdRequestUi, updateClawdRequestUi, clearClawdRequestUi, + applyClawdDurableTaskToUi, + mutateHistoryData, workflowContext, messages, onRequireLogin, diff --git a/lib/db/queries.ts b/lib/db/queries.ts index 03c06ed41..b952bcbf2 100644 --- a/lib/db/queries.ts +++ b/lib/db/queries.ts @@ -373,6 +373,30 @@ export async function saveMessages({ messages }: { messages: DBMessage[] }) { } } +export async function deleteMessagesByIds({ + chatId, + messageIds, +}: { + chatId: string; + messageIds: string[]; +}) { + const normalizedIds = Array.from( + new Set(messageIds.map((id) => String(id || "").trim()).filter(Boolean)) + ); + if (normalizedIds.length === 0) return []; + + try { + return await db + .delete(message) + .where( + and(eq(message.chatId, chatId), inArray(message.id, normalizedIds)) + ) + .returning({ id: message.id }); + } catch (_error) { + throw new ChatSDKError("bad_request:database", "Failed to delete messages"); + } +} + export async function updateMessageContent({ messageId, chatId, diff --git a/lib/openclaw-chat/task-state.ts b/lib/openclaw-chat/task-state.ts new file mode 100644 index 000000000..20e382f33 --- /dev/null +++ b/lib/openclaw-chat/task-state.ts @@ -0,0 +1,120 @@ +export const CLAWD_TASK_KIND = "clawd_chat"; + +export type ClawdTaskStatus = + | "queued" + | "running" + | "completed" + | "failed" + | "canceled"; + +export type ClawdTaskProgressPayload = Record; + +export type ClawdTaskMetadata = { + composerMode: "clawd"; + source: "openclaw"; + taskKind: typeof CLAWD_TASK_KIND; + status: ClawdTaskStatus; + userMessageId: string; + assistantMessageId: string; + processingHint: string; + progressEvents: ClawdTaskProgressPayload[]; + skillTagId?: string | null; + skillTagName?: string | null; + requestId?: string | null; + error?: string | null; + createdAt: number; + updatedAt: number; +}; + +const CLAWD_TASK_STATUSES = new Set([ + "queued", + "running", + "completed", + "failed", + "canceled", +]); + +export function normalizeClawdTaskMetadata( + value: unknown +): ClawdTaskMetadata | null { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const raw = value as Record; + if (raw.taskKind !== CLAWD_TASK_KIND || raw.composerMode !== "clawd") { + return null; + } + + const status = String(raw.status || "") as ClawdTaskStatus; + const userMessageId = String(raw.userMessageId || "").trim(); + const assistantMessageId = String(raw.assistantMessageId || "").trim(); + if ( + !CLAWD_TASK_STATUSES.has(status) || + !userMessageId || + !assistantMessageId + ) { + return null; + } + + const createdAt = Number(raw.createdAt); + const updatedAt = Number(raw.updatedAt); + return { + composerMode: "clawd", + source: "openclaw", + taskKind: CLAWD_TASK_KIND, + status, + userMessageId, + assistantMessageId, + processingHint: String(raw.processingHint || "").trim(), + progressEvents: Array.isArray(raw.progressEvents) + ? raw.progressEvents.filter( + (item): item is ClawdTaskProgressPayload => + Boolean(item) && typeof item === "object" && !Array.isArray(item) + ) + : [], + skillTagId: raw.skillTagId ? String(raw.skillTagId) : null, + skillTagName: raw.skillTagName ? String(raw.skillTagName) : null, + requestId: raw.requestId ? String(raw.requestId) : null, + error: raw.error ? String(raw.error) : null, + createdAt: Number.isFinite(createdAt) ? createdAt : Date.now(), + updatedAt: Number.isFinite(updatedAt) ? updatedAt : Date.now(), + }; +} + +export function isPendingClawdTaskStatus(status: ClawdTaskStatus) { + return status === "queued" || status === "running"; +} + +export function appendClawdTaskProgress( + current: ClawdTaskProgressPayload[], + progress: ClawdTaskProgressPayload, + maxItems = 200 +) { + const eventId = String(progress.event_id || "").trim(); + const sequence = String(progress.sequence || "").trim(); + const key = eventId + ? `event:${eventId}` + : sequence + ? `sequence:${sequence}` + : [ + String(progress.step_code || ""), + String(progress.step_state || ""), + String(progress.message || progress.display || ""), + ].join("|"); + const next = current.some((item) => { + const itemEventId = String(item.event_id || "").trim(); + const itemSequence = String(item.sequence || "").trim(); + const itemKey = itemEventId + ? `event:${itemEventId}` + : itemSequence + ? `sequence:${itemSequence}` + : [ + String(item.step_code || ""), + String(item.step_state || ""), + String(item.message || item.display || ""), + ].join("|"); + return itemKey === key; + }) + ? current + : [...current, progress]; + + return next.slice(-Math.max(1, maxItems)); +} diff --git a/lib/openclaw-chat/task-store.ts b/lib/openclaw-chat/task-store.ts new file mode 100644 index 000000000..651498505 --- /dev/null +++ b/lib/openclaw-chat/task-store.ts @@ -0,0 +1,258 @@ +import "server-only"; + +import { + deleteMessagesByIds, + getMessageById, + getMessagesByChatId, + saveMessages, + updateMessageById, +} from "@/lib/db/queries"; +import type { DBMessage } from "@/lib/db/schema"; +import { + appendClawdTaskProgress, + CLAWD_TASK_KIND, + type ClawdTaskMetadata, + type ClawdTaskProgressPayload, + type ClawdTaskStatus, + normalizeClawdTaskMetadata, +} from "@/lib/openclaw-chat/task-state"; + +type PendingClawdTaskParams = { + chatId: string; + userMessageId: string; + assistantMessageId: string; + displayMessage: string; + attachments: DBMessage["attachments"]; + spreadsheetAttachments: unknown[]; + skillTagId: string; + skillTagName: string; + processingHint: string; +}; + +function buildTaskMetadata(params: { + userMessageId: string; + assistantMessageId: string; + status: ClawdTaskStatus; + processingHint: string; + skillTagId?: string | null; + skillTagName?: string | null; + progressEvents?: ClawdTaskProgressPayload[]; + requestId?: string | null; + error?: string | null; + createdAt?: number; + updatedAt?: number; +}): ClawdTaskMetadata { + const now = Date.now(); + return { + composerMode: "clawd", + source: "openclaw", + taskKind: CLAWD_TASK_KIND, + status: params.status, + userMessageId: params.userMessageId, + assistantMessageId: params.assistantMessageId, + processingHint: params.processingHint, + progressEvents: params.progressEvents || [], + skillTagId: params.skillTagId || null, + skillTagName: params.skillTagName || null, + requestId: params.requestId || null, + error: params.error || null, + createdAt: params.createdAt || now, + updatedAt: params.updatedAt || now, + }; +} + +export async function persistPendingClawdTask( + params: PendingClawdTaskParams +) { + const existingRows = await getMessageById({ id: params.userMessageId }); + const existing = existingRows.find((row) => row.chatId === params.chatId); + if (existing) { + return normalizeClawdTaskMetadata(existing.videoMetadata); + } + + const existingMessages = await getMessagesByChatId({ id: params.chatId }).catch( + () => [] + ); + const latestTimestamp = existingMessages.reduce((latest, item) => { + const timestamp = item.createdAt?.getTime?.() || 0; + return Math.max(latest, timestamp); + }, 0); + const userCreatedAt = new Date(Math.max(Date.now(), latestTimestamp + 1)); + const assistantCreatedAt = new Date(userCreatedAt.getTime() + 1); + const taskMetadata = buildTaskMetadata({ + userMessageId: params.userMessageId, + assistantMessageId: params.assistantMessageId, + status: "queued", + processingHint: params.processingHint, + skillTagId: params.skillTagId, + skillTagName: params.skillTagName, + }); + + await saveMessages({ + messages: [ + { + id: params.userMessageId, + chatId: params.chatId, + role: "user", + parts: [{ type: "text", text: params.displayMessage }], + attachments: params.attachments, + createdAt: userCreatedAt, + videoStorageStatus: null, + temporaryVideoUrl: null, + permanentVideoUrl: null, + archivedAt: null, + videoMetadata: { + ...taskMetadata, + spreadsheetAttachmentCount: params.spreadsheetAttachments.length, + spreadsheetAttachments: params.spreadsheetAttachments, + }, + }, + { + id: params.assistantMessageId, + chatId: params.chatId, + role: "assistant", + parts: [], + attachments: [], + createdAt: assistantCreatedAt, + videoStorageStatus: null, + temporaryVideoUrl: null, + permanentVideoUrl: null, + archivedAt: null, + videoMetadata: taskMetadata, + }, + ], + }); + + return taskMetadata; +} + +async function updateClawdAssistantTask(params: { + chatId: string; + assistantMessageId: string; + status?: ClawdTaskStatus; + progress?: ClawdTaskProgressPayload; + parts?: DBMessage["parts"]; + requestId?: string | null; + error?: string | null; +}) { + const rows = await getMessageById({ id: params.assistantMessageId }); + const message = rows.find((row) => row.chatId === params.chatId); + if (!message) return null; + const metadata = normalizeClawdTaskMetadata(message.videoMetadata); + if (!metadata) return null; + + const progressEvents = params.progress + ? appendClawdTaskProgress(metadata.progressEvents, params.progress) + : metadata.progressEvents; + const nextMetadata: ClawdTaskMetadata = { + ...metadata, + status: params.status || metadata.status, + progressEvents, + requestId: + params.requestId !== undefined ? params.requestId : metadata.requestId, + error: params.error !== undefined ? params.error : metadata.error, + updatedAt: Date.now(), + }; + await updateMessageById({ + messageId: params.assistantMessageId, + chatId: params.chatId, + parts: params.parts, + videoMetadata: nextMetadata, + }); + return nextMetadata; +} + +export async function markClawdTaskRunning(params: { + chatId: string; + assistantMessageId: string; +}) { + return updateClawdAssistantTask({ ...params, status: "running" }); +} + +export async function appendPersistedClawdTaskProgress(params: { + chatId: string; + assistantMessageId: string; + progress: ClawdTaskProgressPayload; +}) { + return updateClawdAssistantTask({ + ...params, + status: "running", + progress: params.progress, + }); +} + +export async function markClawdTaskCompleted(params: { + chatId: string; + assistantMessageId: string; + answer: string; + requestId?: string | null; +}) { + return updateClawdAssistantTask({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + status: "completed", + parts: [{ type: "text", text: params.answer }], + requestId: params.requestId, + error: null, + }); +} + +export async function markClawdTaskFailed(params: { + chatId: string; + assistantMessageId: string; + error: string; +}) { + return updateClawdAssistantTask({ + ...params, + status: "failed", + parts: [{ type: "text", text: params.error }], + error: params.error, + }); +} + +export async function getPersistedClawdTask(params: { + chatId: string; + userMessageId?: string | null; +}) { + const messages = await getMessagesByChatId({ id: params.chatId }); + const candidates = messages + .filter((message) => message.role === "assistant") + .map((message) => ({ + message, + metadata: normalizeClawdTaskMetadata(message.videoMetadata), + })) + .filter( + ( + item + ): item is { message: DBMessage; metadata: ClawdTaskMetadata } => + Boolean(item.metadata) && + (!params.userMessageId || + item.metadata?.userMessageId === params.userMessageId) + ) + .sort( + (left, right) => + right.message.createdAt.getTime() - left.message.createdAt.getTime() + ); + const latest = candidates[0]; + if (!latest) return null; + const userMessage = messages.find( + (message) => message.id === latest.metadata.userMessageId + ); + return { + metadata: latest.metadata, + assistantMessage: latest.message, + userMessage: userMessage || null, + }; +} + +export async function deletePersistedClawdTask(params: { + chatId: string; + userMessageId: string; +}) { + const task = await getPersistedClawdTask(params); + const messageIds = [params.userMessageId]; + if (task?.metadata.assistantMessageId) { + messageIds.push(task.metadata.assistantMessageId); + } + return deleteMessagesByIds({ chatId: params.chatId, messageIds }); +} diff --git a/playwright.helpers.config.ts b/playwright.helpers.config.ts index 215aabfc6..24e01a8fb 100644 --- a/playwright.helpers.config.ts +++ b/playwright.helpers.config.ts @@ -47,6 +47,8 @@ export default defineConfig({ /routes\/video-subtitle-erase-tools\.test\.ts/, /routes\/home-sidebar-auth\.test\.ts/, /routes\/openclaw-client-errors\.test\.ts/, + /routes\/openclaw-durable-task-state\.test\.ts/, + /routes\/openclaw-request-state\.test\.ts/, /routes\/openclaw-video-attachments\.test\.ts/, ], fullyParallel: true, diff --git a/tests/routes/openclaw-durable-task-state.test.ts b/tests/routes/openclaw-durable-task-state.test.ts new file mode 100644 index 000000000..2478c8671 --- /dev/null +++ b/tests/routes/openclaw-durable-task-state.test.ts @@ -0,0 +1,45 @@ +import { expect, test } from "@playwright/test"; +import { + appendClawdTaskProgress, + isPendingClawdTaskStatus, + normalizeClawdTaskMetadata, +} from "@/lib/openclaw-chat/task-state"; + +test.describe("OpenClaw durable task state", () => { + test("restores a persisted running task after the page remounts", () => { + const task = normalizeClawdTaskMetadata({ + composerMode: "clawd", + source: "openclaw", + taskKind: "clawd_chat", + status: "running", + userMessageId: "user-message-id", + assistantMessageId: "assistant-message-id", + processingHint: "正在处理", + progressEvents: [{ event_id: 1, message: "请求已接收" }], + createdAt: 100, + updatedAt: 200, + }); + + expect(task).not.toBeNull(); + expect(task?.userMessageId).toBe("user-message-id"); + expect(task && isPendingClawdTaskStatus(task.status)).toBe(true); + expect(task?.progressEvents).toEqual([ + { event_id: 1, message: "请求已接收" }, + ]); + }); + + test("deduplicates replayed progress events while keeping new steps", () => { + const first = { event_id: 8, message: "任务已恢复" }; + const second = { event_id: 9, message: "正在选择工具" }; + let events = appendClawdTaskProgress([], first); + events = appendClawdTaskProgress(events, first); + events = appendClawdTaskProgress(events, second); + + expect(events).toEqual([first, second]); + }); + + test("does not classify completed tasks as pending", () => { + expect(isPendingClawdTaskStatus("completed")).toBe(false); + expect(isPendingClawdTaskStatus("failed")).toBe(false); + }); +});