diff --git a/app/api/canvas/clawd-chat/route.ts b/app/api/canvas/clawd-chat/route.ts index 513fc3bc6..fe66d0cfe 100644 --- a/app/api/canvas/clawd-chat/route.ts +++ b/app/api/canvas/clawd-chat/route.ts @@ -576,7 +576,7 @@ export async function POST(request: Request) { : "image/jpeg"), })); - await persistPendingClawdTask({ + const persistedTask = await persistPendingClawdTask({ chatId: parsed.id, userMessageId, assistantMessageId, @@ -587,6 +587,22 @@ export async function POST(request: Request) { skillTagName, processingHint, }); + if ( + persistedTask && + persistedTask.assistantMessageId !== assistantMessageId + ) { + return NextResponse.json( + buildTaskResponse({ + chatId: parsed.id, + userMessageId, + assistantMessageId: persistedTask.assistantMessageId, + status: persistedTask.status, + processingHint: + persistedTask.processingHint || DEFAULT_PROCESSING_HINT, + }), + { status: 202, headers: { "Retry-After": "2" } } + ); + } const userProfile = await getUserById({ id: session.user.id }).catch( () => null diff --git a/lib/openclaw-chat/task-store.ts b/lib/openclaw-chat/task-store.ts index f24c9468a..30f609e2f 100644 --- a/lib/openclaw-chat/task-store.ts +++ b/lib/openclaw-chat/task-store.ts @@ -8,7 +8,10 @@ import { updateMessageById, } from "@/lib/db/queries"; import type { DBMessage } from "@/lib/db/schema"; -import { encryptOpenClawTaskSecret } from "@/lib/openclaw-chat/task-secret"; +import { + decryptOpenClawTaskSecret, + encryptOpenClawTaskSecret, +} from "@/lib/openclaw-chat/task-secret"; import { appendClawdTaskProgress, CLAWD_TASK_KIND, @@ -37,6 +40,32 @@ export type ClawdUpstreamTaskState = { lastEventId?: number | null; }; +export async function findPersistedClawdUpstreamTaskState(params: { + chatId: string; + taskId: string; +}): Promise { + const messages = await getMessagesByChatId({ id: params.chatId }).catch( + () => [] + ); + for (const message of messages.toReversed()) { + const metadata = normalizeClawdTaskMetadata(message.videoMetadata); + if ( + !metadata || + metadata.upstreamTaskId !== params.taskId || + !metadata.upstreamTaskKey + ) { + continue; + } + return { + taskId: metadata.upstreamTaskId, + taskKey: decryptOpenClawTaskSecret(metadata.upstreamTaskKey), + eventsEndpoint: metadata.upstreamEventsEndpoint, + lastEventId: metadata.upstreamLastEventId, + }; + } + return null; +} + function buildTaskMetadata(params: { userMessageId: string; assistantMessageId: string; diff --git a/lib/openclaw-chat/upstream-task-client.ts b/lib/openclaw-chat/upstream-task-client.ts index 1f132596b..fc8c4a77d 100644 --- a/lib/openclaw-chat/upstream-task-client.ts +++ b/lib/openclaw-chat/upstream-task-client.ts @@ -2,6 +2,7 @@ import "server-only"; import { appendPersistedClawdTaskProgress, + findPersistedClawdUpstreamTaskState, type ClawdUpstreamTaskState, persistClawdUpstreamTaskState, } from "@/lib/openclaw-chat/task-store"; @@ -301,19 +302,37 @@ export async function submitOpenClawDurableTask(params: { const task = readRecord(root.task); const taskId = readText(root.task_id) || readText(task.task_id); if (!taskId) throw new Error("OpenClaw 未返回 task_id"); + const responseTaskKey = readText(root.task_key); + const persistedTask = responseTaskKey + ? null + : await findPersistedClawdUpstreamTaskState({ + chatId: params.target.chatId, + taskId, + }); + const taskKey = responseTaskKey || persistedTask?.taskKey || null; + if (!taskKey) { + throw new OpenClawTaskCredentialUnavailableError(taskId); + } const submittedTask = mergeUpstreamTaskState( { taskId, - taskKey: readText(root.task_key) || null, - eventsEndpoint: getTaskEventsEndpoint(payload, taskId), - lastEventId: 0, + taskKey, + eventsEndpoint: + getTaskEventsEndpoint(payload, taskId) || + persistedTask?.eventsEndpoint || + null, + lastEventId: persistedTask?.lastEventId || 0, }, payload ); // A task-creation response can include the latest upstream cursor even though // none of its events have been persisted locally yet. Always consume the // event stream from the beginning after accepting a new durable task. - const upstreamTask = { ...submittedTask, lastEventId: 0 }; + const upstreamTask = { + ...submittedTask, + taskKey, + lastEventId: persistedTask?.lastEventId || 0, + }; await persistClawdUpstreamTaskState({ ...params.target, upstreamTask, @@ -342,6 +361,11 @@ export async function refreshOpenClawDurableTask(params: { target: ClawdTaskPersistenceTarget; signal?: AbortSignal; }): Promise { + if (!params.upstreamTask.taskKey) { + throw new OpenClawTaskCredentialUnavailableError( + params.upstreamTask.taskId + ); + } const eventsUrl = buildUpstreamUrl( params.config.baseUrl, params.upstreamTask.eventsEndpoint || @@ -459,6 +483,14 @@ export class OpenClawTaskDetachedError extends Error { } } +export class OpenClawTaskCredentialUnavailableError extends OpenClawTaskDetachedError { + constructor(readonly taskId: string) { + super(); + this.name = "OpenClawTaskCredentialUnavailableError"; + this.message = "OpenClaw task key is unavailable; keep the task detached"; + } +} + export async function waitForOpenClawDurableTask(params: { config: OpenClawUpstreamConfig; upstreamTask: ClawdUpstreamTaskState;