From 7d67bd10594645f6c5f09f6c44299bb810b8f976 Mon Sep 17 00:00:00 2001 From: 134 Date: Tue, 11 Aug 2026 11:54:35 +0800 Subject: [PATCH 1/3] fix(claw): resume durable long-running tasks --- .../settings/admin-settings-client.tsx | 36 +- app/api/canvas/clawd-chat/cancel/route.ts | 50 +- app/api/canvas/clawd-chat/route.ts | 411 ++++---------- app/api/canvas/clawd-chat/status/route.ts | 95 +++- app/api/messages/route.ts | 11 +- lib/openclaw-chat/cancel-registry.ts | 17 +- lib/openclaw-chat/config.ts | 19 +- lib/openclaw-chat/task-secret.ts | 67 +++ lib/openclaw-chat/task-state.ts | 13 + lib/openclaw-chat/task-store.ts | 62 +++ lib/openclaw-chat/upstream-task-client.ts | 514 ++++++++++++++++++ tests/routes/openclaw-config.test.ts | 8 + .../openclaw-durable-task-state.test.ts | 6 + 13 files changed, 964 insertions(+), 345 deletions(-) create mode 100644 lib/openclaw-chat/task-secret.ts create mode 100644 lib/openclaw-chat/upstream-task-client.ts diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index 20a8c15ff..d10212934 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -102,7 +102,10 @@ import { } from "@/components/canvas/canvas-skill-icons"; import { HomeAnimationEmbed } from "@/components/home/home-animation-embed"; import { + DEFAULT_OPENCLAW_TASK_TIMEOUT_MS, DEFAULT_OPENCLAW_CHAT_BASE_URL, + MAX_OPENCLAW_TASK_TIMEOUT_MS, + MIN_OPENCLAW_TASK_TIMEOUT_MS, OPENCLAW_CHAT_CONFIG_KEY, type OpenClawChatConfig, } from "@/lib/openclaw-chat/config"; @@ -2241,7 +2244,7 @@ export function AdminSettingsClient() { enabled: false, baseUrl: DEFAULT_OPENCLAW_CHAT_BASE_URL, token: "", - timeoutMs: 600_000, + timeoutMs: DEFAULT_OPENCLAW_TASK_TIMEOUT_MS, }); const [openClawChatTokenConfigured, setOpenClawChatTokenConfigured] = useState(false); @@ -2534,8 +2537,11 @@ export function AdminSettingsClient() { String(raw.baseUrl || "").trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL, token: "", timeoutMs: Math.min( - 600_000, - Math.max(5_000, Math.round(Number(raw.timeoutMs) || 600_000)) + MAX_OPENCLAW_TASK_TIMEOUT_MS, + Math.max( + MIN_OPENCLAW_TASK_TIMEOUT_MS, + Math.round(Number(raw.timeoutMs) || DEFAULT_OPENCLAW_TASK_TIMEOUT_MS) + ) ), }); setOpenClawChatTokenConfigured( @@ -4443,8 +4449,14 @@ export function AdminSettingsClient() { openClawChatConfig.baseUrl.trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL, token: openClawChatConfig.token.trim(), timeoutMs: Math.min( - 600_000, - Math.max(5_000, Math.round(Number(openClawChatConfig.timeoutMs) || 600_000)) + MAX_OPENCLAW_TASK_TIMEOUT_MS, + Math.max( + MIN_OPENCLAW_TASK_TIMEOUT_MS, + Math.round( + Number(openClawChatConfig.timeoutMs) || + DEFAULT_OPENCLAW_TASK_TIMEOUT_MS + ) + ) ), }; @@ -12817,21 +12829,27 @@ export function AdminSettingsClient() {
setOpenClawChatConfig((prev) => ({ ...prev, - timeoutMs: Number(event.target.value) || 600000, + timeoutMs: + Number(event.target.value) || + DEFAULT_OPENCLAW_TASK_TIMEOUT_MS, })) } + step={60_000} type="number" value={openClawChatConfig.timeoutMs} /> +

+ 网络断开会自动续接，该时间只限制业务任务的后台等待周期。 +

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 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", () => { From bbd0795fdcbe48c0313b3c8d768ea1f70ab71619 Mon Sep 17 00:00:00 2001 From: 134 Date: Tue, 11 Aug 2026 13:47:13 +0800 Subject: [PATCH 2/3] fix(image-tools): restore dedicated processing fallbacks --- app/api/enhance/route.ts | 137 ++++++++++++++++++++++++++++ app/api/watermark-remove/route.ts | 145 ++++++++++++++++++++++++------ lib/ai/extend/service.ts | 120 ++++++++++++++++++++++++- 3 files changed, 373 insertions(+), 29 deletions(-) diff --git a/app/api/enhance/route.ts b/app/api/enhance/route.ts index 2f998684b..ea888b293 100644 --- a/app/api/enhance/route.ts +++ b/app/api/enhance/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { createMediaAssetRelation, + getAiProvidersByType, getMediaAssetsByUserAndUrls, getSystemSetting, upsertMediaAsset, @@ -10,6 +11,7 @@ import { uploadFile } from "@/lib/storage/service"; import { db } from "@/lib/db"; import { generatedImage } from "@/lib/db/schema"; import { executeConfiguredVip98kTask } from "@/lib/ai/vip98k-runtime"; +import { createFalClient } from "@fal-ai/client"; // 变清晰API配置类型 type EnhanceApiConfig = { @@ -319,6 +321,59 @@ async function processWithRealESRGANBase64( }; } +async function processWithFalEsrgan( + imageBase64: string, + requestedScaleFactor: number, +): Promise<{ imageBase64: string; processingTime: number }> { + const providers = await getAiProvidersByType({ + type: "fal-image", + enabledOnly: true, + }); + const provider = providers.find((item) => item.apiKey?.trim()); + if (!provider?.apiKey) { + throw new Error("未配置可用的 FAL 专用高清服务"); + } + + const scale = Math.max(2, Math.min(8, requestedScaleFactor || 4)); + const model = scale <= 2 ? "RealESRGAN_x2plus" : "RealESRGAN_x4plus"; + const credentials = provider.apiKey + .trim() + .replace(/^(?:Key|Bearer)\s+/i, ""); + const fal = createFalClient({ credentials }); + const startedAt = Date.now(); + const result: any = await fal.subscribe("fal-ai/esrgan", { + input: { + image_url: `data:image/jpeg;base64,${imageBase64}`, + scale, + model, + output_format: "png", + }, + logs: true, + timeout: 180_000, + }); + const outputUrl = result?.data?.image?.url || result?.image?.url; + if (!outputUrl) { + throw new Error("FAL ESRGAN 专用高清服务未返回图片"); + } + + const outputResponse = await fetch(outputUrl, { + cache: "no-store", + signal: AbortSignal.timeout(120_000), + }); + if (!outputResponse.ok) { + throw new Error( + `FAL ESRGAN 高清结果下载失败（HTTP ${outputResponse.status}）`, + ); + } + + return { + imageBase64: Buffer.from(await outputResponse.arrayBuffer()).toString( + "base64", + ), + processingTime: (Date.now() - startedAt) / 1000, + }; +} + async function processWithVip98kEnhance(params: { userId: string; imageBase64: string; @@ -634,6 +689,42 @@ async function finalizeEnhanceTaskResult(params: { }; } +async function executeFalEsrganFallback(params: { + userId: string; + imageBase64: string; + prompt: string; + chatId: string | null; + messageId: string | null; + sourceImageUrl: string | null; + requestedScaleFactor?: number; + reason: string; +}) { + const scaleFactor = params.requestedScaleFactor || 4; + console.warn("[Enhance-FAL-ESRGAN] 启用专用高清备用链路", { + reason: params.reason.slice(0, 500), + scaleFactor, + }); + const result = await processWithFalEsrgan(params.imageBase64, scaleFactor); + const payload = await finalizeEnhanceTaskResult({ + userId: params.userId, + taskResult: { + status: 200, + data: { image_base64: result.imageBase64 }, + }, + prompt: params.prompt || "图片变清晰处理", + chatId: params.chatId, + messageId: params.messageId, + sourceImageUrls: params.sourceImageUrl + ? [params.sourceImageUrl] + : undefined, + }); + return { + ...payload, + processing_time: result.processingTime, + api_type: "fal-esrgan", + }; +} + export async function GET(request: NextRequest) { try { const session = await auth(); @@ -1026,6 +1117,21 @@ export async function POST(request: NextRequest) { } catch (error) { console.error("[Enhance-RealESRGAN] ❌ 处理失败:", error); + try { + const fallbackPayload = await executeFalEsrganFallback({ + userId, + imageBase64, + prompt, + chatId, + messageId, + sourceImageUrl, + requestedScaleFactor, + reason: error instanceof Error ? error.message : "RealESRGAN 备用服务失败", + }); + return NextResponse.json(fallbackPayload); + } catch (falError) { + console.error("[Enhance-FAL-ESRGAN] 专用高清备用链路失败:", falError); + } return NextResponse.json( { error: "Real-ESRGAN变清晰处理失败", @@ -1180,6 +1286,22 @@ export async function POST(request: NextRequest) { enhanceUrl, processingTime, }); + + try { + const fallbackPayload = await executeFalEsrganFallback({ + userId, + imageBase64, + prompt, + chatId, + messageId, + sourceImageUrl, + requestedScaleFactor, + reason: fetchError instanceof Error ? fetchError.message : "AOS 高清任务创建失败", + }); + return NextResponse.json(fallbackPayload); + } catch (falError) { + console.error("[Enhance-FAL-ESRGAN] 专用高清备用链路失败:", falError); + } // 检查是否是网络连接错误 if (fetchError instanceof Error) { @@ -1246,6 +1368,21 @@ export async function POST(request: NextRequest) { return buildEnhanceTaskPendingResponse(taskId); } const detail = pollError instanceof Error ? pollError.message : "轮询失败"; + try { + const fallbackPayload = await executeFalEsrganFallback({ + userId, + imageBase64, + prompt, + chatId, + messageId, + sourceImageUrl, + requestedScaleFactor, + reason: detail, + }); + return NextResponse.json(fallbackPayload); + } catch (falError) { + console.error("[Enhance-FAL-ESRGAN] 专用高清备用链路失败:", falError); + } const friendlyMessage = getAosFriendlyErrorMessage(detail); return NextResponse.json( { error: "变清晰处理失败", detail: friendlyMessage || detail }, diff --git a/app/api/watermark-remove/route.ts b/app/api/watermark-remove/route.ts index b22886db2..2c1227008 100644 --- a/app/api/watermark-remove/route.ts +++ b/app/api/watermark-remove/route.ts @@ -12,6 +12,10 @@ import { } from "@/lib/db/queries"; import { prepareFeaturePointsCharge } from "@/lib/points/feature-charge"; import { getImageToolCreditCost } from "@/lib/points/config"; +import { generateCanvasImage } from "@/lib/canvas/generate-image"; +import { FAL_NANO_BANANA_2_EDIT_MODEL_ID } from "@/lib/ai/fal-image-models"; + +export const maxDuration = 600; // PicWish API 需要公网可访问的 HTTP/HTTPS URL const WATERMARK_SYNC_WAIT_TIMEOUT_MS = 20_000; @@ -115,11 +119,10 @@ function buildWatermarkTaskPendingResponse(taskId: string) { ); } -async function finalizeWatermarkTaskResult(params: { - taskId: string; +async function persistWatermarkResult(params: { + processedImageUrl: string; userId: string; startedAt?: number; - apiKey: string; sourceImageUrls?: string[]; }): Promise<{ success: boolean; @@ -127,17 +130,8 @@ async function finalizeWatermarkTaskResult(params: { thumbnail_url: string; processing_time?: number; }> { - const pollResult = await pollForResult(params.taskId, params.apiKey, WATERMARK_SYNC_WAIT_TIMEOUT_MS); - if (pollResult.pending) { - throw new Error("WATERMARK_TASK_PENDING"); - } - if (!pollResult.success || !pollResult.imageUrl) { - throw new Error(pollResult.error || "去水印处理失败"); - } - - const processedImageUrl = pollResult.imageUrl; try { - const imageResponse = await fetch(processedImageUrl); + const imageResponse = await fetch(params.processedImageUrl); if (!imageResponse.ok) { throw new Error("下载处理后的图片失败"); } @@ -217,7 +211,7 @@ async function finalizeWatermarkTaskResult(params: { } } } catch (assetError) { - console.error("[WatermarkRemove] ⚠️ MediaAsset 注册失败:", assetError); + console.error("[WatermarkRemove] MediaAsset 注册失败:", assetError); } } @@ -232,14 +226,85 @@ async function finalizeWatermarkTaskResult(params: { console.error("[WatermarkRemove] 上传结果失败，回退原始URL:", uploadError); return { success: true, - original_url: processedImageUrl, - thumbnail_url: processedImageUrl, + original_url: params.processedImageUrl, + thumbnail_url: params.processedImageUrl, processing_time: typeof params.startedAt === "number" ? (Date.now() - params.startedAt) / 1000 : undefined, }; } } +function shouldUseFalWatermarkFallback(status: number, detail: string) { + return ( + status === 401 || + status === 402 || + status === 403 || + status === 429 || + status >= 500 || + /quota|credit|balance|exceed|expired|invalid|unauthor/i.test(detail) + ); +} + +async function removeWatermarkWithFal(params: { + imageUrl: string; + userId: string; + startedAt: number; + sourceImageUrls?: string[]; +}) { + console.warn("[WatermarkRemove] AOS unavailable; using FAL Nano Banana 2 edit fallback"); + const result = await generateCanvasImage({ + prompt: [ + "仅移除图片中后期叠加的水印、平台角标和覆盖式文字标记。", + "必须完整保留原图主体、商品、背景、构图、尺寸、颜色、光影和清晰度。", + "用周围真实纹理自然补全被遮挡区域，不得新增文字、标志或其他物体。", + ].join(""), + model: FAL_NANO_BANANA_2_EDIT_MODEL_ID, + aspectRatio: "auto", + imageSize: "2K", + imageCount: 1, + referenceImages: [params.imageUrl], + allowFailover: false, + debugLabel: "watermark-remove-fal-fallback", + }); + const processedImageUrl = result.images[0]?.originalUrl; + if (!processedImageUrl) { + throw new Error("FAL 去水印未返回可用图片"); + } + return persistWatermarkResult({ + processedImageUrl, + userId: params.userId, + startedAt: params.startedAt, + sourceImageUrls: params.sourceImageUrls, + }); +} + +async function finalizeWatermarkTaskResult(params: { + taskId: string; + userId: string; + startedAt?: number; + apiKey: string; + sourceImageUrls?: string[]; +}): Promise<{ + success: boolean; + original_url: string; + thumbnail_url: string; + processing_time?: number; +}> { + const pollResult = await pollForResult(params.taskId, params.apiKey, WATERMARK_SYNC_WAIT_TIMEOUT_MS); + if (pollResult.pending) { + throw new Error("WATERMARK_TASK_PENDING"); + } + if (!pollResult.success || !pollResult.imageUrl) { + throw new Error(pollResult.error || "去水印处理失败"); + } + return persistWatermarkResult({ + processedImageUrl: pollResult.imageUrl, + userId: params.userId, + startedAt: params.startedAt, + sourceImageUrls: params.sourceImageUrls, + }); +} + export async function GET(request: NextRequest) { try { const session = await auth(); @@ -306,13 +371,6 @@ export async function POST(request: NextRequest) { // 获取API配置 const config = await getWatermarkRemoveConfig(); - if (!config?.apiKey) { - return NextResponse.json( - { error: "去水印服务未配置，请联系管理员在系统设置中配置API密钥" }, - { status: 500 } - ); - } - console.log("[WatermarkRemove] 使用原图URL:", image_url.substring(0, 100) + "..."); // 处理图片URL，确保获取公网可访问的URL @@ -396,6 +454,24 @@ export async function POST(request: NextRequest) { return NextResponse.json(pointsCharge.payload, { status: pointsCharge.status }); } + const sourceImageUrls = + typeof image_url === "string" && /^https?:\/\//i.test(image_url) + ? [image_url] + : undefined; + + if (!config?.apiKey) { + const fallbackResult = await removeWatermarkWithFal({ + imageUrl: publicImageUrl, + userId: session.user.id, + startedAt: startTime, + sourceImageUrls, + }); + if (pointsCharge.chargeable) { + await pointsCharge.consume({ note: "FAL 去水印备用链路消耗积分" }); + } + return NextResponse.json(fallbackResult); + } + // 创建去水印任务 const formData = new FormData(); formData.append("sync", "0"); @@ -415,8 +491,24 @@ export async function POST(request: NextRequest) { if (!createResponse.ok) { const errorText = await createResponse.text(); console.error("[WatermarkRemove] 创建任务失败:", errorText); + if (shouldUseFalWatermarkFallback(createResponse.status, errorText)) { + try { + const fallbackResult = await removeWatermarkWithFal({ + imageUrl: publicImageUrl, + userId: session.user.id, + startedAt: startTime, + sourceImageUrls, + }); + if (pointsCharge.chargeable) { + await pointsCharge.consume({ note: "FAL 去水印备用链路消耗积分" }); + } + return NextResponse.json(fallbackResult); + } catch (fallbackError) { + console.error("[WatermarkRemove] FAL 备用链路失败:", fallbackError); + } + } return NextResponse.json( - { error: "创建去水印任务失败" }, + { error: "去水印服务暂时不可用，请稍后重试" }, { status: 500 } ); } @@ -448,10 +540,7 @@ export async function POST(request: NextRequest) { userId: session.user.id, startedAt: startTime, apiKey: config.apiKey, - sourceImageUrls: - typeof image_url === "string" && /^https?:\/\//i.test(image_url) - ? [image_url] - : undefined, + sourceImageUrls, }); return NextResponse.json(result); } catch (error) { 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( From 398b2ec83c0cdea76e5070b6b9e2462ea1ad29f2 Mon Sep 17 00:00:00 2001 From: 134 Date: Tue, 11 Aug 2026 14:04:35 +0800 Subject: [PATCH 3/3] feat(batch): sync dedicated image tools --- components/scene-fusion-view.tsx | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/components/scene-fusion-view.tsx b/components/scene-fusion-view.tsx index cbca236f9..f33e15432 100644 --- a/components/scene-fusion-view.tsx +++ b/components/scene-fusion-view.tsx @@ -8,6 +8,7 @@ import { Image as ImageIcon, Layers, Loader2, + Maximize2, Plus, RefreshCw, Sparkles, @@ -97,6 +98,7 @@ type BatchMode = | "random-bg" | "rmbg" | "enhance" + | "extend" | "watermark"; type ItemStatus = "pending" | "processing" | "completed" | "failed"; @@ -336,6 +338,12 @@ const modeOptions: Array<{ description: "多图同步增强画质", icon: Sparkles, }, + { + id: "extend", + label: "批量图像拓展", + description: "多图同步扩展画面", + icon: Maximize2, + }, { id: "watermark", label: "批量去水印", @@ -356,6 +364,8 @@ function getModeActionLabel(mode: BatchMode, count: number) { return `批量抠图 (${count} 张)`; case "enhance": return `批量变清晰 (${count} 张)`; + case "extend": + return `批量图像拓展 (${count} 张)`; case "watermark": return `批量去水印 (${count} 张)`; } @@ -367,6 +377,8 @@ function getModePrompt(mode: BatchMode) { return "批量处理-抠图"; case "enhance": return "批量处理-变清晰"; + case "extend": + return "批量处理-图像拓展"; case "watermark": return "批量处理-去水印"; default: @@ -1771,6 +1783,25 @@ export function SceneFusionView({ }; } + if (mode === "extend") { + const payload = await requestLongTaskResult({ + endpoint: "/api/extend", + payload: { + image_url: item.image, + prompt: getModePrompt(mode), + includeBase64: false, + }, + }); + if (!payload?.success) { + throw new Error(payload?.error || payload?.detail || "图像拓展失败"); + } + return { + result: payload.thumbnail_url || payload.original_url || item.image, + originalUrl: + payload.original_url || payload.thumbnail_url || item.image, + }; + } + if (mode === "watermark") { const payload = await requestLongTaskResult({ endpoint: "/api/watermark-remove", @@ -2798,6 +2829,7 @@ export function SceneFusionView({ {(mode === "rmbg" || mode === "enhance" || + mode === "extend" || mode === "watermark") && (