From a68f9832eda69317a56800e5aaca421bd1495c4a 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 | 1012 +++++++++-------- app/api/canvas/clawd-chat/status/route.ts | 101 ++ components/canvas/CanvasChatPanel.tsx | 789 ++++++++++--- .../canvas/chat-panel/clawd-request-state.ts | 60 + 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 + 10 files changed, 1804 insertions(+), 619 deletions(-) create mode 100644 app/api/canvas/clawd-chat/status/route.ts create mode 100644 components/canvas/chat-panel/clawd-request-state.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 31c2d1d67..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,24 +24,30 @@ import { OPENCLAW_UPGRADE_RETRY_MESSAGE, } from "@/lib/openclaw-chat/errors"; import { - buildClawdCancelKey, - 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(""), @@ -67,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 对话"; @@ -82,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 { @@ -115,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({ @@ -168,204 +200,437 @@ 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; } } -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 parseOpenClawBackgroundTask( + payload: unknown +): OpenClawBackgroundTask | null { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) { + return null; } + const rawPayload = payload as Record; + const rawTask = + rawPayload.background_task && + typeof rawPayload.background_task === "object" && + !Array.isArray(rawPayload.background_task) + ? (rawPayload.background_task as Record) + : null; + if (!rawTask) return null; - 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] }); + const jobId = String(rawTask.job_id || "").trim(); + const state = String(rawTask.state || "").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 parsedPercent = Number(rawTask.progress_percent); 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, + jobId, + state, + stage: String(rawTask.stage || "").trim(), + phase: String(rawTask.phase || "").trim(), + displayText: String( + rawTask.display_text || rawTask.phase || rawPayload.answer || "" + ).trim(), + updatedAt: String(rawTask.updated_at || "").trim(), + completed, + total, + 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) + ), + maxWaitSeconds: Math.max( + 1, + Math.min(25, Number(rawTask.max_wait_seconds) || 20) + ), + }; +} + +function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) { + const message = + task.displayText || + task.phase || + (task.state === "processing" ? "任务正在处理" : "任务状态已更新"); + return { + step_code: task.stage || "background_task", + step_state: + task.state === "failed" + ? "failed" + : task.state === "processing" + ? "started" + : "completed", + message, + display: message, + completed: task.completed, + ...(task.total != null ? { total: task.total } : {}), + ...(task.progressPercent != null + ? { progress_percent: task.progressPercent } + : {}), + stage: task.stage, + state: task.state, + updated_at: task.updatedAt, + }; +} + +function buildOpenClawTaskStatusUrl(params: { + baseUrl: string; + task: OpenClawBackgroundTask; + userId: string; + conversationId: string; +}) { + const base = new URL(params.baseUrl); + const endpoint = params.task.pollEndpoint.startsWith("/") + ? params.task.pollEndpoint + : "/api/chat/task-status"; + const url = new URL(endpoint, base.origin); + url.searchParams.set("job_id", params.task.jobId); + url.searchParams.set("user_id", params.userId); + url.searchParams.set("conversation_id", params.conversationId); + url.searchParams.set("wait_seconds", String(params.task.maxWaitSeconds)); + if (params.task.updatedAt) { + url.searchParams.set("known_updated_at", params.task.updatedAt); + } + return url; +} + +function getAnswer(payload: unknown) { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) { + return ""; + } + return String((payload as Record).answer || "").trim(); +} + +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; + controller: AbortController; + chatId: string; + assistantMessageId: string; + userId: string; + conversationId: string; +}) { + let task = params.initialTask; + let finalPayload = params.initialPayload; + let failureCount = 0; + + await appendPersistedClawdTaskProgress({ + chatId: params.chatId, + assistantMessageId: params.assistantMessageId, + progress: buildOpenClawTaskProgress(task), + }); + + 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, + }), + { + 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: "任务仍在运行，正在重新连接进度服务", }, - attachments: userMessage.attachments, - videoMetadata: userMessage.videoMetadata, + }); + await new Promise((resolve) => setTimeout(resolve, task.pollAfterMs)); + } + } + return finalPayload; +} + +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", }, - { - id: assistantMessage.id, - role: "assistant", - parts: assistantMessage.parts, - createdAt: assistantMessage.createdAt.toISOString(), + 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: { - createdAt: assistantMessage.createdAt.toISOString(), - composerMode: "clawd", + 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, }, - videoMetadata: assistantMessage.videoMetadata, - }, - ], + 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( { @@ -392,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 } @@ -411,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< @@ -465,10 +733,6 @@ export async function POST(request: Request) { }; }); - if (isClawdRequestCanceled(cancelKey)) { - return createCanceledResponse(); - } - if (!existingChat) { await saveChat({ id: parsed.id, @@ -479,242 +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 response = await fetch(endpoint, { - method: "POST", - headers: { - Authorization: `Bearer ${config.token}`, - "Content-Type": "application/json", - }, - 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); - 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 } - ); - } - } 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 dd3a53b8c..70a2b4dc5 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -298,6 +298,13 @@ import { stripCanvasModelRoutingLines, stripMediaMarkup, } from "@/components/canvas/chat-panel/canvas-chat-media"; +import { + clearClawdRequestState, + startClawdRequestState, + type ClawdRequestState, + type ClawdRequestStateByChatId, + updateClawdRequestState, +} from "@/components/canvas/chat-panel/clawd-request-state"; import { buildOpenClawVideoDisplayText, buildOpenClawVideoReferenceText, @@ -568,6 +575,200 @@ type PendingClawdRequestRestoreSnapshot = { shouldRestoreSkillTag: boolean; }; +type ClawdProgressEvent = { + key: string; + eventId?: number; + sequence?: number; + stepCode: string; + stepState: string; + message: string; + currentOrderId: string; + completed?: number; + total?: number; + 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 { + if (progress.heartbeat === true) return null; + const message = String( + progress.message || progress.display || "" + ).trim(); + if (!message) return null; + const eventId = Number(progress.event_id); + const sequence = Number(progress.sequence); + const stepCode = String(progress.step_code || "").trim(); + const stepState = String(progress.step_state || "").trim(); + const currentOrderId = String(progress.current_order_id || "").trim(); + const completed = Number(progress.completed); + const total = Number(progress.total); + const progressPercent = Number(progress.progress_percent); + const normalizedEventId = Number.isFinite(eventId) && eventId > 0 + ? eventId + : undefined; + const normalizedSequence = Number.isFinite(sequence) && sequence > 0 + ? sequence + : undefined; + + return { + key: + normalizedEventId != null + ? `event-${normalizedEventId}` + : `sequence-${normalizedSequence || 0}-${stepCode}-${message}`, + eventId: normalizedEventId, + sequence: normalizedSequence, + stepCode, + stepState, + message, + currentOrderId, + completed: + Number.isFinite(completed) && completed >= 0 ? completed : undefined, + total: + progress.total != null && Number.isFinite(total) && total > 0 + ? total + : undefined, + progressPercent: + progress.progress_percent != null && + Number.isFinite(progressPercent) && + progressPercent >= 0 && + progressPercent <= 100 + ? Math.round(progressPercent) + : undefined, + }; +} + function isClawdSubmittedComposerText( snapshot: ClawdComposerSubmissionSnapshot, value: string @@ -668,64 +869,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"; @@ -1998,30 +2141,97 @@ function CanvasAssistantTypingPlaceholder({ label = "分析中", onCancel, cancelLabel = "停止", + progressEvents = [], }: { hint?: string | null; label?: string; onCancel?: () => void; cancelLabel?: string; + progressEvents?: ClawdProgressEvent[]; }) { + const visibleProgressEvents = progressEvents.slice(-12); + const hiddenProgressCount = Math.max( + 0, + progressEvents.length - visibleProgressEvents.length + ); return (
-
- {hint ? ( -
+
+ {visibleProgressEvents.length > 0 ? ( +
+
+ 执行进度 + {progressEvents.length} 步 +
+ {hiddenProgressCount > 0 ? ( +
+ 已记录此前 {hiddenProgressCount} 步 +
+ ) : null} +
+ {visibleProgressEvents.map((progressEvent, index) => { + const isLatest = index === visibleProgressEvents.length - 1; + const isFailed = progressEvent.stepState === "failed"; + const isActive = + isLatest && + ["accepted", "queued", "started", "waiting"].includes( + progressEvent.stepState + ); + return ( +
+ {isFailed ? ( + + ) : isActive ? ( + + ) : ( + + )} +
+
{progressEvent.message}
+ {progressEvent.currentOrderId || progressEvent.total ? ( +
+ {progressEvent.currentOrderId + ? `订单 ${progressEvent.currentOrderId}` + : ""} + {progressEvent.currentOrderId && progressEvent.total + ? " · " + : ""} + {progressEvent.total + ? `${progressEvent.completed || 0}/${progressEvent.total}${ + progressEvent.progressPercent != null + ? ` · ${progressEvent.progressPercent}%` + : "" + }` + : ""} +
+ ) : null} +
+
+ ); + })} +
+
+ ) : hint ? ( +
{hint}
) : null}
- + {label} {onCancel ? (