diff --git a/lib/ai/extend/service.ts b/lib/ai/extend/service.ts
index 76c2d238c..f2189e236 100644
--- a/lib/ai/extend/service.ts
+++ b/lib/ai/extend/service.ts
@@ -18,6 +18,16 @@ type ExtendConfig = {
const VIP98K_EXPANSION_FALLBACK_MODEL = "google/nano-banana-pro/edit";
const VIP98K_SOURCE_DOWNLOAD_TIMEOUT_MS = 120_000;
+const DASHSCOPE_MIN_IMAGE_EDGE = 512;
+const DASHSCOPE_MAX_IMAGE_EDGE = 4096;
+const DASHSCOPE_MAX_IMAGE_BYTES = 10 * 1024 * 1024;
+const DASHSCOPE_SUPPORTED_IMAGE_FORMATS = new Set([
+ "jpeg",
+ "jpg",
+ "png",
+ "heif",
+ "webp",
+]);
export interface ExecuteExtendJobParams {
userId: string;
@@ -126,6 +136,105 @@ async function uploadImageToS3ForAPI(imageBase64: string): Promise {
}
}
+async function normalizeDashscopeExpansionInput(imageUrl: string): Promise {
+ const response = await fetch(imageUrl, {
+ cache: "no-store",
+ redirect: "follow",
+ signal: AbortSignal.timeout(VIP98K_SOURCE_DOWNLOAD_TIMEOUT_MS),
+ });
+ if (!response.ok) {
+ throw new ExtendServiceError(
+ "下载待拓展图片失败",
+ 502,
+ `${response.status} ${response.statusText}`,
+ );
+ }
+
+ const sourceBuffer = Buffer.from(await response.arrayBuffer());
+ const sharp = await import("sharp");
+ const image = sharp.default(sourceBuffer, { failOn: "none" });
+ const metadata = await image.metadata();
+ const width = Number(metadata.width || 0);
+ const height = Number(metadata.height || 0);
+ const format = String(metadata.format || "").toLowerCase();
+ if (!width || !height) {
+ throw new ExtendServiceError("无法识别待拓展图片尺寸", 422);
+ }
+
+ const withinEdgeLimits =
+ width >= DASHSCOPE_MIN_IMAGE_EDGE &&
+ height >= DASHSCOPE_MIN_IMAGE_EDGE &&
+ width <= DASHSCOPE_MAX_IMAGE_EDGE &&
+ height <= DASHSCOPE_MAX_IMAGE_EDGE;
+ const withinFileLimit = sourceBuffer.length <= DASHSCOPE_MAX_IMAGE_BYTES;
+ if (
+ withinEdgeLimits &&
+ withinFileLimit &&
+ DASHSCOPE_SUPPORTED_IMAGE_FORMATS.has(format)
+ ) {
+ return imageUrl;
+ }
+
+ const downscale = Math.min(
+ 1,
+ DASHSCOPE_MAX_IMAGE_EDGE / width,
+ DASHSCOPE_MAX_IMAGE_EDGE / height,
+ );
+ let targetWidth = Math.max(1, Math.round(width * downscale));
+ let targetHeight = Math.max(1, Math.round(height * downscale));
+ if (Math.min(targetWidth, targetHeight) < DASHSCOPE_MIN_IMAGE_EDGE) {
+ const upscale =
+ DASHSCOPE_MIN_IMAGE_EDGE / Math.min(targetWidth, targetHeight);
+ targetWidth = Math.max(1, Math.round(targetWidth * upscale));
+ targetHeight = Math.max(1, Math.round(targetHeight * upscale));
+ }
+
+ const resizeScale = Math.min(
+ 1,
+ DASHSCOPE_MAX_IMAGE_EDGE / targetWidth,
+ DASHSCOPE_MAX_IMAGE_EDGE / targetHeight,
+ );
+ targetWidth = Math.max(1, Math.round(targetWidth * resizeScale));
+ targetHeight = Math.max(1, Math.round(targetHeight * resizeScale));
+
+ const canvasWidth = Math.max(DASHSCOPE_MIN_IMAGE_EDGE, targetWidth);
+ const canvasHeight = Math.max(DASHSCOPE_MIN_IMAGE_EDGE, targetHeight);
+ const left = Math.floor((canvasWidth - targetWidth) / 2);
+ const top = Math.floor((canvasHeight - targetHeight) / 2);
+ const normalizedBuffer = await image
+ .rotate()
+ .resize(targetWidth, targetHeight, { fit: "fill" })
+ .flatten({ background: "#ffffff" })
+ .extend({
+ left,
+ right: canvasWidth - targetWidth - left,
+ top,
+ bottom: canvasHeight - targetHeight - top,
+ background: "#ffffff",
+ })
+ .jpeg({ quality: 88, mozjpeg: true })
+ .toBuffer();
+
+ const normalizedArrayBuffer = normalizedBuffer.buffer.slice(
+ normalizedBuffer.byteOffset,
+ normalizedBuffer.byteOffset + normalizedBuffer.byteLength,
+ ) as ArrayBuffer;
+ const uploaded = await uploadFile(
+ `extend-dashscope-input-${Date.now()}-${Math.random().toString(36).slice(2, 8)}.jpg`,
+ normalizedArrayBuffer,
+ );
+ console.info("[Extend Service] Normalized DashScope input image", {
+ sourceWidth: width,
+ sourceHeight: height,
+ sourceBytes: sourceBuffer.length,
+ sourceFormat: format,
+ targetWidth: canvasWidth,
+ targetHeight: canvasHeight,
+ targetBytes: normalizedBuffer.length,
+ });
+ return uploaded.url;
+}
+
function isTemporaryVip98kExpansionError(error: unknown) {
const message = error instanceof Error ? error.message : String(error || "");
return /HTTP\s+(?:429|5\d\d)|temporar(?:y|ily)|timeout|timed out|unavailable/i.test(
@@ -365,8 +474,17 @@ export async function executeExtendJob(
throw new ExtendServiceError("无法获取输入图片URL", 400);
}
+ const preparedInputImageUrl =
+ config.provider === "vip98k"
+ ? inputImageUrl
+ : await normalizeDashscopeExpansionInput(inputImageUrl);
const startTime = Date.now();
- const extendedImageUrl = await extendImage(inputImageUrl, config, xScale, yScale);
+ const extendedImageUrl = await extendImage(
+ preparedInputImageUrl,
+ config,
+ xScale,
+ yScale,
+ );
const imageResponse = await fetch(extendedImageUrl);
if (!imageResponse.ok) {
throw new ExtendServiceError(
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 1b9d6604f..57cfdc8ba 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", () => {