diff --git a/app/api/canvas/clawd-chat/cancel/route.ts b/app/api/canvas/clawd-chat/cancel/route.ts
index 970b4ed52..72818b5de 100644
--- a/app/api/canvas/clawd-chat/cancel/route.ts
+++ b/app/api/canvas/clawd-chat/cancel/route.ts
@@ -1,7 +1,7 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { auth } from "@/app/(auth)/auth";
-import { getChatById } from "@/lib/db/queries";
+import { getChatById, getSystemSetting } from "@/lib/db/queries";
import {
buildClawdCancelKey,
cancelClawdRequest,
@@ -10,7 +10,18 @@ import {
OPENCLAW_AUTH_REQUIRED_CODE,
OPENCLAW_AUTH_REQUIRED_MESSAGE,
} from "@/lib/openclaw-chat/errors";
-import { deletePersistedClawdTask } from "@/lib/openclaw-chat/task-store";
+import {
+ normalizeOpenClawChatConfig,
+ OPENCLAW_CHAT_CONFIG_KEY,
+} from "@/lib/openclaw-chat/config";
+import {
+ deletePersistedClawdTask,
+ getPersistedClawdTask,
+} from "@/lib/openclaw-chat/task-store";
+import {
+ getPersistedUpstreamTaskState,
+ stopOpenClawDurableTask,
+} from "@/lib/openclaw-chat/upstream-task-client";
const cancelSchema = z.object({
id: z.string().uuid(),
@@ -39,13 +50,34 @@ export async function POST(request: Request) {
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
}
- const active = cancelClawdRequest(
- buildClawdCancelKey({
- userId: session.user.id,
- chatId: parsed.data.id,
- userMessageId: parsed.data.userMessageId,
- })
- );
+ const cancelKey = buildClawdCancelKey({
+ userId: session.user.id,
+ chatId: parsed.data.id,
+ userMessageId: parsed.data.userMessageId,
+ });
+ const task = await getPersistedClawdTask({
+ chatId: parsed.data.id,
+ userMessageId: parsed.data.userMessageId,
+ }).catch(() => null);
+ const active = cancelClawdRequest(cancelKey);
+ const upstreamTask = task
+ ? getPersistedUpstreamTaskState(task.metadata)
+ : null;
+ if (upstreamTask) {
+ const setting = await getSystemSetting({
+ key: OPENCLAW_CHAT_CONFIG_KEY,
+ }).catch(() => null);
+ const config = normalizeOpenClawChatConfig(setting?.value);
+ if (config.enabled && config.baseUrl && config.token) {
+ await stopOpenClawDurableTask({ config, upstreamTask }).catch((error) => {
+ console.warn("[ClawdChat Cancel] Upstream stop deferred:", {
+ chatId: parsed.data.id,
+ taskId: upstreamTask.taskId,
+ error,
+ });
+ });
+ }
+ }
const deleted = await deletePersistedClawdTask({
chatId: parsed.data.id,
diff --git a/app/api/canvas/clawd-chat/route.ts b/app/api/canvas/clawd-chat/route.ts
index 5e6904e9c..b20345b12 100644
--- a/app/api/canvas/clawd-chat/route.ts
+++ b/app/api/canvas/clawd-chat/route.ts
@@ -37,11 +37,17 @@ import {
import {
appendPersistedClawdTaskProgress,
getPersistedClawdTask,
+ markClawdTaskCanceled,
markClawdTaskCompleted,
markClawdTaskFailed,
markClawdTaskRunning,
persistPendingClawdTask,
} from "@/lib/openclaw-chat/task-store";
+import {
+ OpenClawTaskDetachedError,
+ submitOpenClawDurableTask,
+ waitForOpenClawDurableTask,
+} from "@/lib/openclaw-chat/upstream-task-client";
import { generateUUID } from "@/lib/utils";
const DEFAULT_PROCESSING_HINT =
@@ -92,21 +98,6 @@ type ClawdSpreadsheetAttachment = {
truncated: boolean;
};
-type OpenClawBackgroundTask = {
- jobId: string;
- state: string;
- stage: string;
- phase: string;
- displayText: string;
- updatedAt: string;
- completed: number;
- total: number | null;
- progressPercent: number | null;
- pollEndpoint: string;
- pollAfterMs: number;
- maxWaitSeconds: number;
-};
-
type ExecuteClawdTaskParams = {
assistantMessageId: string;
cancelKey: string;
@@ -237,121 +228,6 @@ async function readOpenClawConfig() {
return normalizeOpenClawChatConfig(setting?.value);
}
-function parseSseBlock(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";
- if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart());
- }
- if (dataLines.length === 0) return null;
- try {
- return { event, data: JSON.parse(dataLines.join("\n")) as unknown };
- } catch {
- return null;
- }
-}
-
-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();
- if (!jobId || !state) return null;
- const completed = Math.max(0, Number(rawTask.completed) || 0);
- const parsedTotal = Number(rawTask.total);
- const total =
- Number.isFinite(parsedTotal) && parsedTotal > 0 ? parsedTotal : null;
- const parsedPercent = Number(rawTask.progress_percent);
-
- 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:
- rawTask.progress_percent != null && Number.isFinite(parsedPercent)
- ? Math.max(0, Math.min(100, parsedPercent))
- : total
- ? Math.round((completed * 100) / total)
- : null,
- pollEndpoint: String(rawTask.poll_endpoint || "").trim(),
- 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)
- ),
- };
-}
-
-function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) {
- const message =
- task.displayText ||
- task.phase ||
- (task.state === "processing" ? "任务正在处理" : "任务状态已更新");
- return {
- step_code: task.stage || "background_task",
- step_state:
- task.state === "failed"
- ? "failed"
- : task.state === "processing"
- ? "started"
- : "completed",
- message,
- display: message,
- completed: task.completed,
- ...(task.total != null ? { total: task.total } : {}),
- ...(task.progressPercent != null
- ? { 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;
-}
-
function getAnswer(payload: unknown) {
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
return "";
@@ -367,145 +243,8 @@ function getRequestId(payload: unknown) {
return value ? String(value) : null;
}
-async function readOpenClawSse(params: {
- response: Response;
- chatId: string;
- assistantMessageId: string;
-}) {
- const reader = params.response.body?.getReader();
- if (!reader) throw new Error("OpenClaw stream is unavailable");
- const decoder = new TextDecoder();
- let buffer = "";
- let finalPayload: Record | null = null;
-
- const processBlock = async (block: string) => {
- const event = parseSseBlock(block);
- if (!event) return;
- if (event.event === "progress") {
- if (event.data && typeof event.data === "object") {
- await appendPersistedClawdTaskProgress({
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- progress: event.data as Record,
- });
- }
- return;
- }
- if (
- event.event === "final" &&
- event.data &&
- typeof event.data === "object" &&
- !Array.isArray(event.data)
- ) {
- finalPayload = event.data as Record;
- }
- if (event.event === "error") {
- throw new Error("OpenClaw stream returned an error");
- }
- };
-
- 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() || "";
- for (const block of blocks) {
- await processBlock(block);
- }
- if (done) break;
- }
- if (buffer.trim()) await processBlock(buffer);
-
- if (!finalPayload) throw new Error("OpenClaw stream ended without a result");
- return finalPayload;
-}
-
-async function pollOpenClawBackgroundTask(params: {
- config: ExecuteClawdTaskParams["config"];
- initialPayload: Record;
- initialTask: OpenClawBackgroundTask;
- controller: AbortController;
- chatId: string;
- assistantMessageId: string;
- userId: string;
- conversationId: string;
-}) {
- let task = params.initialTask;
- let finalPayload = params.initialPayload;
- let failureCount = 0;
-
- await appendPersistedClawdTaskProgress({
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- progress: buildOpenClawTaskProgress(task),
- });
-
- while (task.state === "processing") {
- if (!task.pollEndpoint) {
- throw new Error("OpenClaw background task has no status endpoint");
- }
- try {
- const response = await fetch(
- buildOpenClawTaskStatusUrl({
- baseUrl: params.config.baseUrl,
- task,
- userId: params.userId,
- conversationId: params.conversationId,
- }),
- {
- headers: {
- Authorization: `Bearer ${params.config.token}`,
- Accept: "application/json",
- },
- cache: "no-store",
- signal: params.controller.signal,
- }
- );
- if (!response.ok) {
- throw new Error(`OpenClaw task status failed: ${response.status}`);
- }
- const payload = (await response.json().catch(() => null)) as
- | Record
- | null;
- const nextTask = parseOpenClawBackgroundTask(payload);
- if (!payload || !nextTask) {
- throw new Error("OpenClaw task status payload is invalid");
- }
- failureCount = 0;
- finalPayload = {
- ...payload,
- request_id:
- payload.request_id || params.initialPayload.request_id || null,
- };
- task = nextTask;
- await appendPersistedClawdTaskProgress({
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- progress: buildOpenClawTaskProgress(task),
- });
- } catch (error) {
- if (params.controller.signal.aborted) throw error;
- failureCount += 1;
- if (failureCount >= 5) throw error;
- await appendPersistedClawdTaskProgress({
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- progress: {
- step_code: task.stage || "background_task",
- step_state: "waiting",
- message: "任务仍在运行，正在重新连接进度服务",
- display: "任务仍在运行，正在重新连接进度服务",
- },
- });
- await new Promise((resolve) => setTimeout(resolve, task.pollAfterMs));
- }
- }
- return finalPayload;
-}
-
async function executeClawdTask(params: ExecuteClawdTaskParams) {
const controller = new AbortController();
- const timeout = setTimeout(() => controller.abort(), params.config.timeoutMs);
registerActiveClawdRequest(
params.cancelKey,
controller,
@@ -519,8 +258,7 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
assistantMessageId: params.assistantMessageId,
});
- const isSensitiveSubmission = Boolean(params.sensitiveInput);
- const upstreamBody = JSON.stringify({
+ const upstreamBody: Record = {
conversation_id: params.conversationId,
user_id: params.userId,
user_name: params.userName,
@@ -560,64 +298,106 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
params.spreadsheetMarkdownTruncated || undefined,
...params.metadata,
},
- stream_progress: !isSensitiveSubmission,
+ stream_progress: false,
sensitive_input: params.sensitiveInput,
- });
- const response = await fetch(`${params.config.baseUrl}/api/chat`, {
- method: "POST",
- headers: {
- Authorization: `Bearer ${params.config.token}`,
- "Content-Type": "application/json",
- Accept: isSensitiveSubmission ? "application/json" : "text/event-stream",
- },
- body: upstreamBody,
- signal: controller.signal,
- cache: "no-store",
- });
+ };
- if (!response.ok) {
- throw new Error(`OpenClaw request failed: ${response.status}`);
- }
-
- const contentType = response.headers.get("content-type") || "";
- let payload = contentType.includes("text/event-stream")
- ? await readOpenClawSse({
- response,
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- })
- : ((await response.json().catch(() => null)) as Record<
+ if (params.sensitiveInput) {
+ const timeout = setTimeout(
+ () =>
+ controller.abort(
+ new DOMException("OpenClaw request timed out", "TimeoutError")
+ ),
+ params.config.timeoutMs
+ );
+ try {
+ const response = await fetch(`${params.config.baseUrl}/api/chat`, {
+ method: "POST",
+ headers: {
+ Authorization: `Bearer ${params.config.token}`,
+ "Content-Type": "application/json",
+ Accept: "application/json",
+ },
+ body: JSON.stringify(upstreamBody),
+ signal: controller.signal,
+ cache: "no-store",
+ });
+ const payload = (await response.json().catch(() => null)) as Record<
string,
unknown
- > | null);
- if (!payload) throw new Error("OpenClaw returned an invalid response");
-
- const backgroundTask = parseOpenClawBackgroundTask(payload);
- if (backgroundTask?.state === "processing") {
- payload = await pollOpenClawBackgroundTask({
- config: params.config,
- initialPayload: payload,
- initialTask: backgroundTask,
- controller,
- chatId: params.chatId,
- assistantMessageId: params.assistantMessageId,
- userId: params.userId,
- conversationId: params.conversationId,
- });
+ > | null;
+ if (!response.ok) {
+ throw new Error(`OpenClaw request failed: ${response.status}`);
+ }
+ if (!payload) throw new Error("OpenClaw returned an invalid response");
+ const answer = getAnswer(payload);
+ if (!answer) throw new Error("OpenClaw returned an empty answer");
+ await markClawdTaskCompleted({
+ chatId: params.chatId,
+ assistantMessageId: params.assistantMessageId,
+ answer,
+ requestId: getRequestId(payload),
+ });
+ clearClawdRequestCancellation(params.cancelKey);
+ return;
+ } finally {
+ clearTimeout(timeout);
+ }
}
- if (isClawdRequestCanceled(params.cancelKey)) return;
- const answer = getAnswer(payload);
- if (!answer) throw new Error("OpenClaw returned an empty answer");
- await markClawdTaskCompleted({
+ const target = {
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
- answer,
- requestId: getRequestId(payload),
+ };
+ const upstreamTask = await submitOpenClawDurableTask({
+ config: params.config,
+ body: upstreamBody,
+ idempotencyKey: `${params.userId}:${params.chatId}:${params.userMessageId}`,
+ target,
+ signal: controller.signal,
});
+ const outcome = await waitForOpenClawDurableTask({
+ config: params.config,
+ upstreamTask,
+ target,
+ deadlineAt: Date.now() + params.config.timeoutMs,
+ signal: controller.signal,
+ });
+
+ if (isClawdRequestCanceled(params.cancelKey)) return;
+ if (outcome.status === "completed") {
+ await markClawdTaskCompleted({
+ ...target,
+ answer: outcome.answer,
+ requestId: outcome.requestId,
+ });
+ } else if (outcome.status === "canceled") {
+ await markClawdTaskCanceled({
+ ...target,
+ message: outcome.error,
+ });
+ } else if (outcome.status === "failed") {
+ await markClawdTaskFailed({
+ ...target,
+ error: outcome.error,
+ });
+ }
clearClawdRequestCancellation(params.cancelKey);
} catch (error) {
if (isClawdRequestCanceled(params.cancelKey)) return;
+ if (error instanceof OpenClawTaskDetachedError) {
+ await appendPersistedClawdTaskProgress({
+ chatId: params.chatId,
+ assistantMessageId: params.assistantMessageId,
+ progress: {
+ step_code: "task.detached",
+ step_state: "waiting",
+ message: "任务仍在 OpenClaw 后台执行，打开原对话后会自动续接结果",
+ display: "任务仍在 OpenClaw 后台执行，打开原对话后会自动续接结果",
+ },
+ }).catch(() => null);
+ return;
+ }
console.error("[ClawdChat API] Durable task failed:", error);
await markClawdTaskFailed({
chatId: params.chatId,
@@ -632,7 +412,6 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
params.sensitiveInput.verification_code = "";
}
params.sensitiveInput = undefined;
- clearTimeout(timeout);
unregisterActiveClawdRequest(params.cancelKey);
}
}
diff --git a/app/api/canvas/clawd-chat/status/route.ts b/app/api/canvas/clawd-chat/status/route.ts
index 81be3d3f1..70c369393 100644
--- a/app/api/canvas/clawd-chat/status/route.ts
+++ b/app/api/canvas/clawd-chat/status/route.ts
@@ -1,12 +1,32 @@
import { NextResponse } from "next/server";
import { z } from "zod";
import { auth } from "@/app/(auth)/auth";
-import { getChatById } from "@/lib/db/queries";
+import { getChatById, getSystemSetting } from "@/lib/db/queries";
+import {
+ buildClawdCancelKey,
+ hasActiveClawdRequest,
+ registerActiveClawdRequest,
+ unregisterActiveClawdRequest,
+} from "@/lib/openclaw-chat/cancel-registry";
+import {
+ normalizeOpenClawChatConfig,
+ OPENCLAW_CHAT_CONFIG_KEY,
+} from "@/lib/openclaw-chat/config";
import {
OPENCLAW_AUTH_REQUIRED_CODE,
OPENCLAW_AUTH_REQUIRED_MESSAGE,
} from "@/lib/openclaw-chat/errors";
-import { getPersistedClawdTask } from "@/lib/openclaw-chat/task-store";
+import {
+ getPersistedClawdTask,
+ markClawdTaskCanceled,
+ markClawdTaskCompleted,
+ markClawdTaskFailed,
+} from "@/lib/openclaw-chat/task-store";
+import { isPendingClawdTaskStatus } from "@/lib/openclaw-chat/task-state";
+import {
+ getPersistedUpstreamTaskState,
+ refreshOpenClawDurableTask,
+} from "@/lib/openclaw-chat/upstream-task-client";
const statusSchema = z.object({
id: z.string().uuid(),
@@ -15,6 +35,64 @@ const statusSchema = z.object({
export const dynamic = "force-dynamic";
export const revalidate = 0;
+export const maxDuration = 60;
+
+async function reconcileDetachedOpenClawTask(params: {
+ chatId: string;
+ userId: string;
+ task: NonNullable>>;
+}) {
+ if (!isPendingClawdTaskStatus(params.task.metadata.status)) return;
+ const upstreamTask = getPersistedUpstreamTaskState(params.task.metadata);
+ if (!upstreamTask) return;
+ const cancelKey = buildClawdCancelKey({
+ userId: params.userId,
+ chatId: params.chatId,
+ userMessageId: params.task.metadata.userMessageId,
+ });
+ if (hasActiveClawdRequest(cancelKey)) return;
+
+ const controller = new AbortController();
+ registerActiveClawdRequest(cancelKey, controller, 30_000);
+ const target = {
+ chatId: params.chatId,
+ assistantMessageId: params.task.metadata.assistantMessageId,
+ };
+ try {
+ const setting = await getSystemSetting({
+ key: OPENCLAW_CHAT_CONFIG_KEY,
+ }).catch(() => null);
+ const config = normalizeOpenClawChatConfig(setting?.value);
+ if (!config.enabled || !config.baseUrl || !config.token) return;
+ const outcome = await refreshOpenClawDurableTask({
+ config,
+ upstreamTask,
+ target,
+ signal: controller.signal,
+ });
+ if (outcome.status === "completed") {
+ await markClawdTaskCompleted({
+ ...target,
+ answer: outcome.answer,
+ requestId: outcome.requestId,
+ });
+ } else if (outcome.status === "failed") {
+ await markClawdTaskFailed({ ...target, error: outcome.error });
+ } else if (outcome.status === "canceled") {
+ await markClawdTaskCanceled({ ...target, message: outcome.error });
+ }
+ } catch (error) {
+ if (!controller.signal.aborted) {
+ console.warn("[ClawdChat Status] Upstream reconciliation deferred:", {
+ chatId: params.chatId,
+ taskId: upstreamTask.taskId,
+ error,
+ });
+ }
+ } finally {
+ unregisterActiveClawdRequest(cancelKey);
+ }
+}
export async function GET(request: Request) {
const session = await auth().catch(() => null);
@@ -45,7 +123,7 @@ export async function GET(request: Request) {
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
}
- const task = await getPersistedClawdTask({
+ let task = await getPersistedClawdTask({
chatId: parsed.data.id,
userMessageId: parsed.data.userMessageId,
});
@@ -56,6 +134,17 @@ export async function GET(request: Request) {
);
}
+ await reconcileDetachedOpenClawTask({
+ chatId: parsed.data.id,
+ userId: session.user.id,
+ task,
+ });
+ task =
+ (await getPersistedClawdTask({
+ chatId: parsed.data.id,
+ userMessageId: parsed.data.userMessageId,
+ })) || task;
+
const userTextPart = Array.isArray(task.userMessage?.parts)
? task.userMessage.parts.find(
(part) =>
diff --git a/app/api/messages/route.ts b/app/api/messages/route.ts
index 8983da8a1..65b4f9114 100644
--- a/app/api/messages/route.ts
+++ b/app/api/messages/route.ts
@@ -32,6 +32,15 @@ const FAL_IMAGE_TASK_POLL_INTERVAL_MS = 2_000;
const falTaskLastPolledAt = new Map();
const falTaskSyncInFlight = new Set();
+function toPublicVideoMetadata(value: DBMessage["videoMetadata"]) {
+ if (!value || typeof value !== "object" || Array.isArray(value)) return value;
+ const metadata = value as Record;
+ if (metadata.taskKind !== "clawd_chat") return value;
+ const publicMetadata = { ...metadata };
+ delete publicMetadata.upstreamTaskKey;
+ return publicMetadata;
+}
+
function createEmptyMessagesResponse() {
return NextResponse.json(
{ messages: [] },
@@ -277,7 +286,7 @@ export async function GET(request: NextRequest) {
createdAt: msg.createdAt,
parts: parts,
attachments: msg.attachments || [],
- videoMetadata: msg.videoMetadata || null,
+ videoMetadata: toPublicVideoMetadata(msg.videoMetadata) || null,
};
});
diff --git a/lib/openclaw-chat/cancel-registry.ts b/lib/openclaw-chat/cancel-registry.ts
index 8896d37be..dfd21cd88 100644
--- a/lib/openclaw-chat/cancel-registry.ts
+++ b/lib/openclaw-chat/cancel-registry.ts
@@ -5,8 +5,16 @@ type ActiveClawdRequest = {
expiresAt: number;
};
-const activeRequests = new Map();
-const canceledRequests = new Map();
+const registry = globalThis as typeof globalThis & {
+ __activeClawdRequests?: Map;
+ __canceledClawdRequests?: Map;
+};
+const activeRequests =
+ registry.__activeClawdRequests || new Map();
+const canceledRequests =
+ registry.__canceledClawdRequests || new Map();
+registry.__activeClawdRequests = activeRequests;
+registry.__canceledClawdRequests = canceledRequests;
function cleanupExpiredClawdRequests(now = Date.now()) {
for (const [key, request] of activeRequests.entries()) {
@@ -50,6 +58,11 @@ export function unregisterActiveClawdRequest(key: string) {
activeRequests.delete(key);
}
+export function hasActiveClawdRequest(key: string) {
+ cleanupExpiredClawdRequests();
+ return activeRequests.has(key);
+}
+
export function cancelClawdRequest(key: string) {
cleanupExpiredClawdRequests();
canceledRequests.set(key, Date.now() + CANCELED_REQUEST_TTL_MS);
diff --git a/lib/openclaw-chat/config.ts b/lib/openclaw-chat/config.ts
index 6fa3c3124..d1ae5ddb9 100644
--- a/lib/openclaw-chat/config.ts
+++ b/lib/openclaw-chat/config.ts
@@ -20,11 +20,12 @@ export type AdminOpenClawChatConfig = Omit & {
tokenPreview: string;
};
-const DEFAULT_TIMEOUT_MS = 30 * 60 * 1000;
-const MIN_TIMEOUT_MS = 5_000;
+export const DEFAULT_OPENCLAW_TASK_TIMEOUT_MS = 30 * 60 * 1000;
+export const MIN_OPENCLAW_TASK_TIMEOUT_MS = 5 * 60 * 1000;
// Complex OpenClaw skills can legitimately run for up to thirty minutes.
// Keep this aligned with the route maxDuration and the integration contract.
-const MAX_TIMEOUT_MS = 30 * 60 * 1000;
+export const MAX_OPENCLAW_TASK_TIMEOUT_MS = 30 * 60 * 1000;
+const LEGACY_CONNECTION_TIMEOUT_MAX_MS = 5 * 60 * 1000;
function readRecord(value: unknown): Record {
return value && typeof value === "object" && !Array.isArray(value)
@@ -41,8 +42,16 @@ function normalizeBaseUrl(value: unknown) {
function normalizeTimeoutMs(value: unknown) {
const numeric = Number(value);
- if (!Number.isFinite(numeric)) return DEFAULT_TIMEOUT_MS;
- return Math.min(MAX_TIMEOUT_MS, Math.max(MIN_TIMEOUT_MS, Math.round(numeric)));
+ if (!Number.isFinite(numeric)) return DEFAULT_OPENCLAW_TASK_TIMEOUT_MS;
+ // Older settings described one SSE connection and were commonly saved as
+ // 180-300 seconds. Durable tasks use a separate business-task lifetime.
+ if (numeric <= LEGACY_CONNECTION_TIMEOUT_MAX_MS) {
+ return DEFAULT_OPENCLAW_TASK_TIMEOUT_MS;
+ }
+ return Math.min(
+ MAX_OPENCLAW_TASK_TIMEOUT_MS,
+ Math.max(MIN_OPENCLAW_TASK_TIMEOUT_MS, Math.round(numeric))
+ );
}
export function normalizeOpenClawChatConfig(
diff --git a/lib/openclaw-chat/task-secret.ts b/lib/openclaw-chat/task-secret.ts
new file mode 100644
index 000000000..1e37e12c6
--- /dev/null
+++ b/lib/openclaw-chat/task-secret.ts
@@ -0,0 +1,67 @@
+import "server-only";
+
+import {
+ createCipheriv,
+ createDecipheriv,
+ createHash,
+ randomBytes,
+} from "node:crypto";
+
+const ENCRYPTED_SECRET_PREFIX = "enc:v1";
+
+function getEncryptionKey() {
+ const secret = String(
+ process.env.OPENCLAW_TASK_SECRET ||
+ process.env.AUTH_SECRET ||
+ process.env.NEXTAUTH_SECRET ||
+ ""
+ ).trim();
+ if (!secret) {
+ throw new Error(
+ "OPENCLAW_TASK_SECRET or AUTH_SECRET is required for durable tasks"
+ );
+ }
+ return createHash("sha256").update(secret, "utf8").digest();
+}
+
+export function encryptOpenClawTaskSecret(value: string) {
+ const normalized = value.trim();
+ if (!normalized || normalized.startsWith(`${ENCRYPTED_SECRET_PREFIX}:`)) {
+ return normalized;
+ }
+ const iv = randomBytes(12);
+ const cipher = createCipheriv("aes-256-gcm", getEncryptionKey(), iv);
+ const encrypted = Buffer.concat([
+ cipher.update(normalized, "utf8"),
+ cipher.final(),
+ ]);
+ const authTag = cipher.getAuthTag();
+ return [
+ ENCRYPTED_SECRET_PREFIX,
+ iv.toString("base64url"),
+ authTag.toString("base64url"),
+ encrypted.toString("base64url"),
+ ].join(":");
+}
+
+export function decryptOpenClawTaskSecret(value: string) {
+ const normalized = value.trim();
+ if (!normalized.startsWith(`${ENCRYPTED_SECRET_PREFIX}:`)) {
+ // Backward compatibility for tasks persisted before encryption was added.
+ return normalized;
+ }
+ const [, , ivValue, authTagValue, encryptedValue] = normalized.split(":");
+ if (!ivValue || !authTagValue || !encryptedValue) {
+ throw new Error("Invalid encrypted OpenClaw task secret");
+ }
+ const decipher = createDecipheriv(
+ "aes-256-gcm",
+ getEncryptionKey(),
+ Buffer.from(ivValue, "base64url")
+ );
+ decipher.setAuthTag(Buffer.from(authTagValue, "base64url"));
+ return Buffer.concat([
+ decipher.update(Buffer.from(encryptedValue, "base64url")),
+ decipher.final(),
+ ]).toString("utf8");
+}
diff --git a/lib/openclaw-chat/task-state.ts b/lib/openclaw-chat/task-state.ts
index 20e382f33..e23292594 100644
--- a/lib/openclaw-chat/task-state.ts
+++ b/lib/openclaw-chat/task-state.ts
@@ -22,6 +22,10 @@ export type ClawdTaskMetadata = {
skillTagName?: string | null;
requestId?: string | null;
error?: string | null;
+ upstreamTaskId?: string | null;
+ upstreamTaskKey?: string | null;
+ upstreamEventsEndpoint?: string | null;
+ upstreamLastEventId?: number | null;
createdAt: number;
updatedAt: number;
};
@@ -56,6 +60,7 @@ export function normalizeClawdTaskMetadata(
const createdAt = Number(raw.createdAt);
const updatedAt = Number(raw.updatedAt);
+ const upstreamLastEventId = Number(raw.upstreamLastEventId);
return {
composerMode: "clawd",
source: "openclaw",
@@ -74,6 +79,14 @@ export function normalizeClawdTaskMetadata(
skillTagName: raw.skillTagName ? String(raw.skillTagName) : null,
requestId: raw.requestId ? String(raw.requestId) : null,
error: raw.error ? String(raw.error) : null,
+ upstreamTaskId: raw.upstreamTaskId ? String(raw.upstreamTaskId) : null,
+ upstreamTaskKey: raw.upstreamTaskKey ? String(raw.upstreamTaskKey) : null,
+ upstreamEventsEndpoint: raw.upstreamEventsEndpoint
+ ? String(raw.upstreamEventsEndpoint)
+ : null,
+ upstreamLastEventId: Number.isFinite(upstreamLastEventId)
+ ? Math.max(0, Math.trunc(upstreamLastEventId))
+ : null,
createdAt: Number.isFinite(createdAt) ? createdAt : Date.now(),
updatedAt: Number.isFinite(updatedAt) ? updatedAt : Date.now(),
};
diff --git a/lib/openclaw-chat/task-store.ts b/lib/openclaw-chat/task-store.ts
index 651498505..f24c9468a 100644
--- a/lib/openclaw-chat/task-store.ts
+++ b/lib/openclaw-chat/task-store.ts
@@ -8,6 +8,7 @@ import {
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,
@@ -29,6 +30,13 @@ type PendingClawdTaskParams = {
processingHint: string;
};
+export type ClawdUpstreamTaskState = {
+ taskId: string;
+ taskKey?: string | null;
+ eventsEndpoint?: string | null;
+ lastEventId?: number | null;
+};
+
function buildTaskMetadata(params: {
userMessageId: string;
assistantMessageId: string;
@@ -39,6 +47,7 @@ function buildTaskMetadata(params: {
progressEvents?: ClawdTaskProgressPayload[];
requestId?: string | null;
error?: string | null;
+ upstreamTask?: ClawdUpstreamTaskState | null;
createdAt?: number;
updatedAt?: number;
}): ClawdTaskMetadata {
@@ -56,6 +65,12 @@ function buildTaskMetadata(params: {
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,
};
@@ -134,6 +149,7 @@ async function updateClawdAssistantTask(params: {
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);
@@ -151,6 +167,22 @@ async function updateClawdAssistantTask(params: {
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({
@@ -173,11 +205,26 @@ 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,
});
}
@@ -210,6 +257,21 @@ export async function markClawdTaskFailed(params: {
});
}
+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;
diff --git a/lib/openclaw-chat/upstream-task-client.ts b/lib/openclaw-chat/upstream-task-client.ts
new file mode 100644
index 000000000..1f132596b
--- /dev/null
+++ b/lib/openclaw-chat/upstream-task-client.ts
@@ -0,0 +1,514 @@
+import "server-only";
+
+import {
+ appendPersistedClawdTaskProgress,
+ type ClawdUpstreamTaskState,
+ persistClawdUpstreamTaskState,
+} from "@/lib/openclaw-chat/task-store";
+import type { ClawdTaskMetadata } from "@/lib/openclaw-chat/task-state";
+import { decryptOpenClawTaskSecret } from "@/lib/openclaw-chat/task-secret";
+
+type OpenClawUpstreamConfig = {
+ baseUrl: string;
+ token: string;
+ timeoutMs: number;
+};
+
+type ClawdTaskPersistenceTarget = {
+ chatId: string;
+ assistantMessageId: string;
+};
+
+export type OpenClawTaskOutcome =
+ | {
+ status: "pending";
+ upstreamTask: ClawdUpstreamTaskState;
+ }
+ | {
+ status: "completed";
+ upstreamTask: ClawdUpstreamTaskState;
+ answer: string;
+ requestId: string | null;
+ }
+ | {
+ status: "failed" | "canceled";
+ upstreamTask: ClawdUpstreamTaskState;
+ error: string;
+ };
+
+const UPSTREAM_REQUEST_TIMEOUT_MS = 15_000;
+const UPSTREAM_POLL_INTERVAL_MS = 1600;
+const TERMINAL_STATUSES = new Set([
+ "succeeded",
+ "completed",
+ "failed",
+ "cancelled",
+ "canceled",
+]);
+
+class OpenClawUpstreamHttpError extends Error {
+ constructor(
+ message: string,
+ readonly status: number
+ ) {
+ super(message);
+ this.name = "OpenClawUpstreamHttpError";
+ }
+}
+
+function readRecord(value: unknown): Record {
+ return value && typeof value === "object" && !Array.isArray(value)
+ ? (value as Record)
+ : {};
+}
+
+function readText(value: unknown) {
+ return typeof value === "string" ? value.trim() : "";
+}
+
+function readEventId(value: unknown) {
+ const parsed = Number(value);
+ return Number.isFinite(parsed) ? Math.max(0, Math.trunc(parsed)) : 0;
+}
+
+function getTaskStatus(payload: unknown) {
+ return readText(readRecord(readRecord(payload).task).status).toLowerCase();
+}
+
+function getTaskEventsEndpoint(payload: unknown, taskId: string) {
+ const task = readRecord(readRecord(payload).task);
+ return (
+ readText(task.events_endpoint) ||
+ readText(readRecord(payload).events_endpoint) ||
+ `/api/chat/tasks/${encodeURIComponent(taskId)}/events`
+ );
+}
+
+function getAnswer(payload: unknown) {
+ const root = readRecord(payload);
+ const output = readRecord(root.output);
+ const result = readRecord(root.result);
+ return (
+ readText(root.answer) || readText(output.answer) || readText(result.answer)
+ );
+}
+
+function getRequestId(payload: unknown) {
+ const root = readRecord(payload);
+ const output = readRecord(root.output);
+ return readText(root.request_id) || readText(output.request_id) || null;
+}
+
+function getTaskError(payload: unknown, fallback: string) {
+ const root = readRecord(payload);
+ const error = readRecord(root.error);
+ const outputError = readRecord(readRecord(root.output).error);
+ return (
+ readText(error.message) ||
+ readText(outputError.message) ||
+ readText(root.message) ||
+ fallback
+ );
+}
+
+function buildUpstreamUrl(baseUrl: string, endpoint: string) {
+ const normalizedBaseUrl = `${baseUrl.replace(/\/+$/, "")}/`;
+ return new URL(endpoint, normalizedBaseUrl);
+}
+
+function buildTaskEndpoint(taskId: string) {
+ return `/api/chat/tasks/${encodeURIComponent(taskId)}`;
+}
+
+function buildTaskControlEndpoint(taskId: string) {
+ return `${buildTaskEndpoint(taskId)}/control`;
+}
+
+function createRequestController(parentSignal?: AbortSignal) {
+ const controller = new AbortController();
+ const abortFromParent = () => controller.abort(parentSignal?.reason);
+ if (parentSignal?.aborted) {
+ abortFromParent();
+ } else {
+ parentSignal?.addEventListener("abort", abortFromParent, { once: true });
+ }
+ const timeout = setTimeout(
+ () => controller.abort(new DOMException("OpenClaw request timed out", "TimeoutError")),
+ UPSTREAM_REQUEST_TIMEOUT_MS
+ );
+ return {
+ signal: controller.signal,
+ cleanup() {
+ clearTimeout(timeout);
+ parentSignal?.removeEventListener("abort", abortFromParent);
+ },
+ };
+}
+
+async function fetchJson(params: {
+ config: OpenClawUpstreamConfig;
+ endpoint: string;
+ method?: "GET" | "POST";
+ body?: Record;
+ taskKey?: string | null;
+ idempotencyKey?: string;
+ signal?: AbortSignal;
+}) {
+ const requestController = createRequestController(params.signal);
+ try {
+ const response = await fetch(
+ buildUpstreamUrl(params.config.baseUrl, params.endpoint),
+ {
+ method: params.method || "GET",
+ headers: {
+ Authorization: `Bearer ${params.config.token}`,
+ Accept: "application/json",
+ ...(params.body ? { "Content-Type": "application/json" } : {}),
+ ...(params.taskKey
+ ? { "X-Dalongxia-Task-Key": params.taskKey }
+ : {}),
+ ...(params.idempotencyKey
+ ? { "Idempotency-Key": params.idempotencyKey }
+ : {}),
+ },
+ body: params.body ? JSON.stringify(params.body) : undefined,
+ cache: "no-store",
+ signal: requestController.signal,
+ }
+ );
+ const payload = (await response.json().catch(() => null)) as unknown;
+ if (!response.ok) {
+ throw new OpenClawUpstreamHttpError(
+ getTaskError(
+ payload,
+ `OpenClaw request failed: ${response.status}`
+ ),
+ response.status
+ );
+ }
+ return payload;
+ } finally {
+ requestController.cleanup();
+ }
+}
+
+function mergeUpstreamTaskState(
+ current: ClawdUpstreamTaskState,
+ payload: unknown
+) {
+ const root = readRecord(payload);
+ const task = readRecord(root.task);
+ const taskId =
+ readText(root.task_id) || readText(task.task_id) || current.taskId;
+ const eventId = Math.max(
+ current.lastEventId || 0,
+ readEventId(root.next_after_event_id),
+ readEventId(root.event_id)
+ );
+ return {
+ taskId,
+ taskKey: readText(root.task_key) || current.taskKey || null,
+ eventsEndpoint:
+ getTaskEventsEndpoint(payload, taskId) || current.eventsEndpoint || null,
+ lastEventId: eventId,
+ } satisfies ClawdUpstreamTaskState;
+}
+
+function normalizeProgressEvent(
+ event: unknown,
+ taskId: string
+): Record | null {
+ const raw = readRecord(event);
+ if (Object.keys(raw).length === 0) return null;
+ const data = readRecord(raw.data);
+ const eventId = readEventId(raw.event_id);
+ const phase = readText(raw.phase);
+ const eventType = readText(raw.event_type);
+ const message =
+ readText(data.message) ||
+ readText(data.display) ||
+ readText(data.display_text) ||
+ readText(raw.display_text) ||
+ phase ||
+ "任务状态已更新";
+ return {
+ ...data,
+ event_id: eventId,
+ task_id: taskId,
+ event_type: eventType,
+ status: readText(raw.status),
+ phase,
+ step_code: readText(data.step_code) || phase || eventType || "task.progress",
+ step_state: readText(data.step_state) || readText(raw.status) || "started",
+ message,
+ display: message,
+ created_at: raw.created_at,
+ };
+}
+
+export function getPersistedUpstreamTaskState(
+ metadata: ClawdTaskMetadata
+): ClawdUpstreamTaskState | null {
+ const taskId = readText(metadata.upstreamTaskId);
+ if (!taskId) return null;
+ return {
+ taskId,
+ taskKey: metadata.upstreamTaskKey
+ ? decryptOpenClawTaskSecret(metadata.upstreamTaskKey)
+ : null,
+ eventsEndpoint:
+ readText(metadata.upstreamEventsEndpoint) ||
+ `/api/chat/tasks/${encodeURIComponent(taskId)}/events`,
+ lastEventId: readEventId(metadata.upstreamLastEventId),
+ };
+}
+
+export async function submitOpenClawDurableTask(params: {
+ config: OpenClawUpstreamConfig;
+ body: Record;
+ idempotencyKey: string;
+ target: ClawdTaskPersistenceTarget;
+ signal?: AbortSignal;
+}) {
+ let payload: unknown;
+ for (let attempt = 1; attempt <= 3; attempt += 1) {
+ try {
+ payload = await fetchJson({
+ config: params.config,
+ endpoint: "/api/chat/tasks",
+ method: "POST",
+ body: params.body,
+ idempotencyKey: params.idempotencyKey,
+ signal: params.signal,
+ });
+ break;
+ } catch (error) {
+ if (params.signal?.aborted) throw error;
+ if (
+ error instanceof OpenClawUpstreamHttpError &&
+ error.status >= 400 &&
+ error.status < 500 &&
+ error.status !== 408 &&
+ error.status !== 429
+ ) {
+ throw error;
+ }
+ if (attempt >= 3) throw error;
+ await waitForDelay(attempt * 500, params.signal);
+ }
+ }
+ const root = readRecord(payload);
+ const task = readRecord(root.task);
+ const taskId = readText(root.task_id) || readText(task.task_id);
+ if (!taskId) throw new Error("OpenClaw 未返回 task_id");
+ const submittedTask = mergeUpstreamTaskState(
+ {
+ taskId,
+ taskKey: readText(root.task_key) || null,
+ eventsEndpoint: getTaskEventsEndpoint(payload, taskId),
+ 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 };
+ await persistClawdUpstreamTaskState({
+ ...params.target,
+ upstreamTask,
+ });
+ return upstreamTask;
+}
+
+export async function stopOpenClawDurableTask(params: {
+ config: OpenClawUpstreamConfig;
+ upstreamTask: ClawdUpstreamTaskState;
+ signal?: AbortSignal;
+}) {
+ return fetchJson({
+ config: params.config,
+ endpoint: buildTaskControlEndpoint(params.upstreamTask.taskId),
+ method: "POST",
+ body: { action: "stop" },
+ taskKey: params.upstreamTask.taskKey,
+ signal: params.signal,
+ });
+}
+
+export async function refreshOpenClawDurableTask(params: {
+ config: OpenClawUpstreamConfig;
+ upstreamTask: ClawdUpstreamTaskState;
+ target: ClawdTaskPersistenceTarget;
+ signal?: AbortSignal;
+}): Promise {
+ const eventsUrl = buildUpstreamUrl(
+ params.config.baseUrl,
+ params.upstreamTask.eventsEndpoint ||
+ `/api/chat/tasks/${encodeURIComponent(params.upstreamTask.taskId)}/events`
+ );
+ eventsUrl.searchParams.set(
+ "after_event_id",
+ String(params.upstreamTask.lastEventId || 0)
+ );
+ const eventsPayload = await fetchJson({
+ config: params.config,
+ endpoint: eventsUrl.toString(),
+ taskKey: params.upstreamTask.taskKey,
+ signal: params.signal,
+ });
+ const mergedUpstreamTask = mergeUpstreamTaskState(
+ params.upstreamTask,
+ eventsPayload
+ );
+ let upstreamTask = {
+ ...mergedUpstreamTask,
+ lastEventId: params.upstreamTask.lastEventId || 0,
+ };
+ const events = Array.isArray(readRecord(eventsPayload).events)
+ ? (readRecord(eventsPayload).events as unknown[])
+ : [];
+ for (const event of events) {
+ const eventId = readEventId(readRecord(event).event_id);
+ upstreamTask = {
+ ...upstreamTask,
+ lastEventId: Math.max(upstreamTask.lastEventId || 0, eventId),
+ };
+ const progress = normalizeProgressEvent(event, upstreamTask.taskId);
+ if (!progress) continue;
+ await appendPersistedClawdTaskProgress({
+ ...params.target,
+ progress,
+ upstreamTask,
+ });
+ }
+ upstreamTask = {
+ ...upstreamTask,
+ lastEventId: Math.max(
+ upstreamTask.lastEventId || 0,
+ readEventId(readRecord(eventsPayload).next_after_event_id)
+ ),
+ };
+ await persistClawdUpstreamTaskState({
+ ...params.target,
+ upstreamTask,
+ });
+
+ const status = getTaskStatus(eventsPayload);
+ if (!TERMINAL_STATUSES.has(status)) {
+ return { status: "pending", upstreamTask };
+ }
+
+ const terminalPayload = await fetchJson({
+ config: params.config,
+ endpoint: buildTaskEndpoint(upstreamTask.taskId),
+ taskKey: upstreamTask.taskKey,
+ signal: params.signal,
+ });
+ upstreamTask = mergeUpstreamTaskState(upstreamTask, terminalPayload);
+ const terminalStatus = getTaskStatus(terminalPayload) || status;
+ if (terminalStatus === "succeeded" || terminalStatus === "completed") {
+ const answer = getAnswer(terminalPayload);
+ if (!answer) {
+ return {
+ status: "failed",
+ upstreamTask,
+ error: "OpenClaw 任务已结束，但未返回有效结果",
+ };
+ }
+ return {
+ status: "completed",
+ upstreamTask,
+ answer,
+ requestId: getRequestId(terminalPayload),
+ };
+ }
+ if (terminalStatus === "cancelled" || terminalStatus === "canceled") {
+ return { status: "canceled", upstreamTask, error: "任务已停止" };
+ }
+ return {
+ status: "failed",
+ upstreamTask,
+ error: getTaskError(terminalPayload, "OpenClaw 任务执行失败"),
+ };
+}
+
+function waitForDelay(delayMs: number, signal?: AbortSignal) {
+ return new Promise((resolve, reject) => {
+ if (signal?.aborted) {
+ reject(signal.reason || new DOMException("Aborted", "AbortError"));
+ return;
+ }
+ let timer: ReturnType;
+ const onAbort = () => {
+ clearTimeout(timer);
+ reject(signal?.reason || new DOMException("Aborted", "AbortError"));
+ };
+ timer = setTimeout(() => {
+ signal?.removeEventListener("abort", onAbort);
+ resolve();
+ }, delayMs);
+ signal?.addEventListener("abort", onAbort, { once: true });
+ });
+}
+
+export class OpenClawTaskDetachedError extends Error {
+ constructor() {
+ super("OpenClaw task continues in the background");
+ this.name = "OpenClawTaskDetachedError";
+ }
+}
+
+export async function waitForOpenClawDurableTask(params: {
+ config: OpenClawUpstreamConfig;
+ upstreamTask: ClawdUpstreamTaskState;
+ target: ClawdTaskPersistenceTarget;
+ deadlineAt: number;
+ signal?: AbortSignal;
+}) {
+ let upstreamTask = params.upstreamTask;
+ let failureCount = 0;
+ while (!params.signal?.aborted && Date.now() < params.deadlineAt) {
+ try {
+ const outcome = await refreshOpenClawDurableTask({
+ config: params.config,
+ upstreamTask,
+ target: params.target,
+ signal: params.signal,
+ });
+ upstreamTask = outcome.upstreamTask;
+ failureCount = 0;
+ if (outcome.status !== "pending") return outcome;
+ } catch (error) {
+ if (params.signal?.aborted) throw error;
+ if (
+ error instanceof OpenClawUpstreamHttpError &&
+ error.status >= 400 &&
+ error.status < 500 &&
+ error.status !== 408 &&
+ error.status !== 429
+ ) {
+ throw error;
+ }
+ failureCount += 1;
+ await appendPersistedClawdTaskProgress({
+ ...params.target,
+ upstreamTask,
+ progress: {
+ step_code: "task.reconnecting",
+ step_state: "waiting",
+ message: "任务仍在 OpenClaw 后台运行，正在重新连接进度服务",
+ display: "任务仍在 OpenClaw 后台运行，正在重新连接进度服务",
+ },
+ });
+ }
+ await waitForDelay(
+ Math.min(5000, UPSTREAM_POLL_INTERVAL_MS + failureCount * 500),
+ params.signal
+ );
+ }
+ if (params.signal?.aborted) {
+ throw params.signal.reason || new DOMException("Aborted", "AbortError");
+ }
+ throw new OpenClawTaskDetachedError();
+}
diff --git a/tests/routes/openclaw-config.test.ts b/tests/routes/openclaw-config.test.ts
index 39e3466e5..f5b611783 100644
--- a/tests/routes/openclaw-config.test.ts
+++ b/tests/routes/openclaw-config.test.ts
@@ -24,4 +24,12 @@ test.describe("OpenClaw chat configuration", () => {
}).timeoutMs
).toBe(30 * 60 * 1000);
});
+
+ test("migrates legacy three-minute connection timeout to task lifetime", () => {
+ expect(
+ normalizeOpenClawChatConfig({
+ timeoutMs: 180_000,
+ }).timeoutMs
+ ).toBe(30 * 60 * 1000);
+ });
});
diff --git a/tests/routes/openclaw-durable-task-state.test.ts b/tests/routes/openclaw-durable-task-state.test.ts
index 2478c8671..53fe356d4 100644
--- a/tests/routes/openclaw-durable-task-state.test.ts
+++ b/tests/routes/openclaw-durable-task-state.test.ts
@@ -16,6 +16,10 @@ test.describe("OpenClaw durable task state", () => {
assistantMessageId: "assistant-message-id",
processingHint: "正在处理",
progressEvents: [{ event_id: 1, message: "请求已接收" }],
+ upstreamTaskId: "task_test_123",
+ upstreamTaskKey: "enc:v1:iv:tag:ciphertext",
+ upstreamEventsEndpoint: "/api/chat/tasks/task_test_123/events",
+ upstreamLastEventId: 18,
createdAt: 100,
updatedAt: 200,
});
@@ -26,6 +30,8 @@ test.describe("OpenClaw durable task state", () => {
expect(task?.progressEvents).toEqual([
{ event_id: 1, message: "请求已接收" },
]);
+ expect(task?.upstreamTaskId).toBe("task_test_123");
+ expect(task?.upstreamLastEventId).toBe(18);
});
test("deduplicates replayed progress events while keeping new steps", () => {