fix(claw): persist and recover background chat tasks

This commit is contained in:
134 2026-08-06 12:28:50 +08:00
parent c2fcf11280
commit a68f9832ed
10 changed files with 1804 additions and 619 deletions

View File

@ -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

View 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",
},
}
);
}

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,60 @@
export type ClawdRequestState<TProgress> = {
userMessageId: string;
hint: string | null;
progressEvents: TProgress[];
};
export type ClawdRequestStateByChatId<TProgress> = Record<
string,
ClawdRequestState<TProgress>
>;
export function startClawdRequestState<TProgress>(
previousState: ClawdRequestStateByChatId<TProgress>,
chatSessionId: string,
userMessageId: string,
hint: string | null
): ClawdRequestStateByChatId<TProgress> {
return {
...previousState,
[chatSessionId]: {
userMessageId,
hint,
progressEvents: [],
},
};
}
export function updateClawdRequestState<TProgress>(
previousState: ClawdRequestStateByChatId<TProgress>,
chatSessionId: string,
userMessageId: string,
updater: (
currentState: ClawdRequestState<TProgress>
) => ClawdRequestState<TProgress>
): ClawdRequestStateByChatId<TProgress> {
const currentState = previousState[chatSessionId];
if (!currentState || currentState.userMessageId !== userMessageId) {
return previousState;
}
return {
...previousState,
[chatSessionId]: updater(currentState),
};
}
export function clearClawdRequestState<TProgress>(
previousState: ClawdRequestStateByChatId<TProgress>,
chatSessionId: string,
userMessageId: string
): ClawdRequestStateByChatId<TProgress> {
const currentState = previousState[chatSessionId];
if (!currentState || currentState.userMessageId !== userMessageId) {
return previousState;
}
const nextState = { ...previousState };
delete nextState[chatSessionId];
return nextState;
}

View File

@ -371,6 +371,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,

View 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));
}

View 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 });
}

View File

@ -16,6 +16,8 @@ export default defineConfig({
/routes\/clawd-request-lifecycle\.test\.ts/,
/routes\/canvas-skill-tag-order\.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/,
/routes\/hicad-embedded-performance\.test\.ts/,
/routes\/hicad-manufacturing-extension\.test\.ts/,

View 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);
});
});