fix(claw): persist and recover background chat tasks
This commit is contained in:
134
parent
b3161d2ebc
commit
92647c3789
@ -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,
|
||||
});
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
101
app/api/canvas/clawd-chat/status/route.ts
Normal file
101
app/api/canvas/clawd-chat/status/route.ts
Normal file
@ -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",
|
||||
},
|
||||
}
|
||||
);
|
||||
}
|
||||
@ -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<string, unknown>[];
|
||||
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<void>((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<string, unknown>
|
||||
): ClawdProgressEvent | null {
|
||||
@ -690,64 +821,6 @@ function shouldShowClawdProcessingHint(
|
||||
);
|
||||
}
|
||||
|
||||
async function readClawdResponse(
|
||||
response: Response,
|
||||
onProgress: (progress: Record<string, unknown>) => 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<string, unknown>);
|
||||
} 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<string, { userMessageId: string; controller: AbortController }>
|
||||
>(new Map());
|
||||
const clawdRecoveryPollersRef = useRef<Map<string, AbortController>>(
|
||||
new Map()
|
||||
);
|
||||
const pendingClawdRequestRestoresRef = useRef<
|
||||
Map<string, PendingClawdRequestRestoreSnapshot>
|
||||
>(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<ClawdDurableTaskSnapshot>
|
||||
| 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;
|
||||
const terminalTask = await waitForClawdDurableTask({
|
||||
rawChatId: requestRawChatId,
|
||||
userMessageId,
|
||||
signal: clawdAbortController.signal,
|
||||
onUpdate: (task) => {
|
||||
applyClawdDurableTaskToUi(requestChatId, task);
|
||||
},
|
||||
});
|
||||
if (!terminalTask) {
|
||||
throw new Error("Claw 任务记录暂时不可用");
|
||||
}
|
||||
setMessages((previousMessages) =>
|
||||
sortCanvasMessagesByCreatedAt([
|
||||
...previousMessages.filter(
|
||||
(message) =>
|
||||
message.id !== userMessageId && !returnedIds.has(message.id)
|
||||
),
|
||||
...returnedMessages,
|
||||
])
|
||||
);
|
||||
if (terminalTask.status === "completed") {
|
||||
setClawdTypewriterMessageIds((previousIds) => ({
|
||||
...previousIds,
|
||||
[terminalTask.assistantMessageId]: true,
|
||||
}));
|
||||
}
|
||||
if (requestRawChatId) {
|
||||
void mutate(
|
||||
buildHistoryMessagesUrl(requestRawChatId, { cacheBust: 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,
|
||||
|
||||
@ -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,
|
||||
|
||||
120
lib/openclaw-chat/task-state.ts
Normal file
120
lib/openclaw-chat/task-state.ts
Normal file
@ -0,0 +1,120 @@
|
||||
export const CLAWD_TASK_KIND = "clawd_chat";
|
||||
|
||||
export type ClawdTaskStatus =
|
||||
| "queued"
|
||||
| "running"
|
||||
| "completed"
|
||||
| "failed"
|
||||
| "canceled";
|
||||
|
||||
export type ClawdTaskProgressPayload = Record<string, unknown>;
|
||||
|
||||
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<ClawdTaskStatus>([
|
||||
"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<string, unknown>;
|
||||
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));
|
||||
}
|
||||
258
lib/openclaw-chat/task-store.ts
Normal file
258
lib/openclaw-chat/task-store.ts
Normal file
@ -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 });
|
||||
}
|
||||
@ -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,
|
||||
|
||||
45
tests/routes/openclaw-durable-task-state.test.ts
Normal file
45
tests/routes/openclaw-durable-task-state.test.ts
Normal file
@ -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);
|
||||
});
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user