diff --git a/app/api/canvas/clawd-chat/route.ts b/app/api/canvas/clawd-chat/route.ts index 9e907db6a..a23c4352e 100644 --- a/app/api/canvas/clawd-chat/route.ts +++ b/app/api/canvas/clawd-chat/route.ts @@ -232,6 +232,126 @@ function parseSseBlock(block: string) { } } +type OpenClawBackgroundTask = { + jobId: string; + state: string; + stage: string; + phase: string; + displayText: string; + updatedAt: string; + completed: number; + total: number; + progressPercent: number; + pollEndpoint: string; + pollAfterMs: number; + maxWaitSeconds: number; + autoNotify: boolean; +}; + +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 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 total = Math.max(1, Number(rawTask.total) || 1); + const progressPercent = Math.max( + 0, + Math.min( + 100, + Number.isFinite(Number(rawTask.progress_percent)) + ? Number(rawTask.progress_percent) + : Math.round((completed * 100) / total) + ) + ); + + return { + 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, + pollEndpoint, + 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) + ), + autoNotify: rawTask.auto_notify === true, + }; +} + +function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) { + return { + step_code: task.stage || "background_task", + step_state: + task.state === "failed" + ? "failed" + : task.state === "processing" + ? "started" + : "completed", + message: + task.displayText || + task.phase || + (task.state === "processing" ? "任务正在处理" : "任务状态已更新"), + display: + task.displayText || + task.phase || + (task.state === "processing" ? "任务正在处理" : "任务状态已更新"), + completed: task.completed, + total: task.total, + 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; +} + async function persistClawdResponse({ assistantMessageId, attachments, @@ -361,6 +481,181 @@ async function persistClawdResponse({ }; } +function streamOpenClawBackgroundTask(params: { + request: Request; + cancelKey: string; + config: { baseUrl: string; token: string; timeoutMs: number }; + initialPayload: Record; + initialTask: OpenClawBackgroundTask; + assistantMessageId: string; + attachments: OpenClawChatMediaAttachment[]; + chatId: string; + conversationId: string; + displayMessage: string; + skillTagId: string; + skillTagName: string; + spreadsheetAttachments: ClawdSpreadsheetAttachment[]; + userId: string; + userMessageId: 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 lastProgressKey = ""; + let consecutiveFailures = 0; + + 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, + total: task.total, + 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, + 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: CLAW_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. + } + } + }, + }); + + return new Response(stream, { + headers: { + "Cache-Control": "no-cache, no-store", + "Content-Type": "text/event-stream; charset=utf-8", + "X-Accel-Buffering": "no", + }, + }); +} + export async function POST(request: Request) { const session = await auth().catch(() => null); @@ -500,6 +795,7 @@ export async function POST(request: Request) { headers: { Authorization: `Bearer ${config.token}`, "Content-Type": "application/json", + Accept: "text/event-stream", }, body: JSON.stringify({ conversation_id: conversationId, @@ -657,6 +953,31 @@ export async function POST(request: Request) { { 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 = diff --git a/components/canvas/CanvasChatPanel.tsx b/components/canvas/CanvasChatPanel.tsx index 1e75817e1..e4eb7eda8 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -494,6 +494,65 @@ 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; +}; + +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: Number.isFinite(total) && total > 0 ? total : undefined, + progressPercent: + Number.isFinite(progressPercent) && + progressPercent >= 0 && + progressPercent <= 100 + ? Math.round(progressPercent) + : undefined, + }; +} + function isClawdSubmittedComposerText( snapshot: ClawdComposerSubmissionSnapshot, value: string @@ -1924,16 +1983,85 @@ 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}
@@ -2493,6 +2621,9 @@ export function CanvasChatPanel({ const [isDetailGenerating, setIsDetailGenerating] = useState(false); const [isClawdSubmitting, setIsClawdSubmitting] = useState(false); const [clawdProcessingHint, setClawdProcessingHint] = useState(null); + const [clawdProgressEvents, setClawdProgressEvents] = useState< + ClawdProgressEvent[] + >([]); const [detailSections, setDetailSections] = useState>({ required: true, strategy: false, @@ -7804,6 +7935,7 @@ export function CanvasChatPanel({ clawdAbortControllerRef.current = null; setIsClawdSubmitting(false); setClawdProcessingHint(null); + setClawdProgressEvents([]); releaseSendGuard(snapshot.chatId); restorePendingClawdRequestToComposer(snapshot); toast.message("已停止处理，可以重新编辑后再发送"); @@ -11527,6 +11659,7 @@ export function CanvasChatPanel({ clawdAbortControllerRef.current = clawdAbortController; pendingClawdRequestRestoreRef.current = clawdRequestRestoreSnapshot; setClawdProcessingHint(nextProcessingHint); + setClawdProgressEvents([]); setIsClawdSubmitting(true); setMessages((previousMessages) => sortCanvasMessagesByCreatedAt([ @@ -11584,6 +11717,19 @@ export function CanvasChatPanel({ if (display) { setClawdProcessingHint(display); } + const normalizedProgress = normalizeClawdProgressEvent(progress); + if (normalizedProgress) { + setClawdProgressEvents((previousEvents) => { + if ( + previousEvents.some( + (item) => item.key === normalizedProgress.key + ) + ) { + return previousEvents; + } + return [...previousEvents, normalizedProgress].slice(-2000); + }); + } }); if (response.status === 499 || responseData?.canceled === true) { canceledClawdMessageIdsRef.current.add(userMessageId); @@ -11713,6 +11859,7 @@ export function CanvasChatPanel({ } setIsClawdSubmitting(false); setClawdProcessingHint(null); + setClawdProgressEvents([]); releaseSendGuard(activeChatId); } return; @@ -15183,6 +15330,7 @@ export function CanvasChatPanel({ hint={clawdProcessingHint} label="处理中" onCancel={handleCancelClawdSubmission} + progressEvents={clawdProgressEvents} /> ) : null}