import "server-only"; import { deleteMessagesByIds, getMessageById, getMessagesByChatId, saveMessages, updateMessageById, } from "@/lib/db/queries"; import type { DBMessage } from "@/lib/db/schema"; import { encryptOpenClawTaskSecret } from "@/lib/openclaw-chat/task-secret"; 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; }; export type ClawdUpstreamTaskState = { taskId: string; taskKey?: string | null; eventsEndpoint?: string | null; lastEventId?: number | null; }; 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; upstreamTask?: ClawdUpstreamTaskState | 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, upstreamTaskId: params.upstreamTask?.taskId || null, upstreamTaskKey: params.upstreamTask?.taskKey ? encryptOpenClawTaskSecret(params.upstreamTask.taskKey) : null, upstreamEventsEndpoint: params.upstreamTask?.eventsEndpoint || null, upstreamLastEventId: params.upstreamTask?.lastEventId ?? 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; upstreamTask?: ClawdUpstreamTaskState; }) { 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, upstreamTaskId: params.upstreamTask?.taskId !== undefined ? params.upstreamTask.taskId : metadata.upstreamTaskId, upstreamTaskKey: params.upstreamTask?.taskKey !== undefined ? encryptOpenClawTaskSecret(params.upstreamTask.taskKey || "") || null : metadata.upstreamTaskKey, upstreamEventsEndpoint: params.upstreamTask?.eventsEndpoint !== undefined ? params.upstreamTask.eventsEndpoint : metadata.upstreamEventsEndpoint, upstreamLastEventId: params.upstreamTask?.lastEventId !== undefined ? params.upstreamTask.lastEventId : metadata.upstreamLastEventId, 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; upstreamTask?: ClawdUpstreamTaskState; }) { return updateClawdAssistantTask({ ...params, status: "running", progress: params.progress, upstreamTask: params.upstreamTask, }); } export async function persistClawdUpstreamTaskState(params: { chatId: string; assistantMessageId: string; upstreamTask: ClawdUpstreamTaskState; }) { return updateClawdAssistantTask({ chatId: params.chatId, assistantMessageId: params.assistantMessageId, status: "running", upstreamTask: params.upstreamTask, }); } 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 markClawdTaskCanceled(params: { chatId: string; assistantMessageId: string; message?: string; }) { const message = params.message || "任务已停止"; return updateClawdAssistantTask({ chatId: params.chatId, assistantMessageId: params.assistantMessageId, status: "canceled", parts: [{ type: "text", text: message }], error: null, }); } 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 }); }