merge: preserve server-only production updates
This commit is contained in:
134
commit
e2a8f2dda3
@ -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";
|
||||
@ -2253,7 +2256,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);
|
||||
@ -2546,8 +2549,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(
|
||||
@ -4465,8 +4471,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
|
||||
)
|
||||
)
|
||||
),
|
||||
};
|
||||
|
||||
@ -12867,21 +12879,27 @@ export function AdminSettingsClient() {
|
||||
</div>
|
||||
<div className="space-y-2">
|
||||
<Label htmlFor="openClawChatTimeoutMs">
|
||||
请求超时（毫秒）
|
||||
任务最长执行时间（毫秒）
|
||||
</Label>
|
||||
<Input
|
||||
id="openClawChatTimeoutMs"
|
||||
min={5000}
|
||||
max={600000}
|
||||
min={MIN_OPENCLAW_TASK_TIMEOUT_MS}
|
||||
max={MAX_OPENCLAW_TASK_TIMEOUT_MS}
|
||||
onChange={(event) =>
|
||||
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}
|
||||
/>
|
||||
<p className="text-muted-foreground text-xs">
|
||||
网络断开会自动续接，该时间只限制业务任务的后台等待周期。
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
<div className="space-y-2">
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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<string, unknown>;
|
||||
const rawTask =
|
||||
rawPayload.background_task &&
|
||||
typeof rawPayload.background_task === "object" &&
|
||||
!Array.isArray(rawPayload.background_task)
|
||||
? (rawPayload.background_task as Record<string, unknown>)
|
||||
: 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<string, unknown> | 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<string, unknown>,
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (
|
||||
event.event === "final" &&
|
||||
event.data &&
|
||||
typeof event.data === "object" &&
|
||||
!Array.isArray(event.data)
|
||||
) {
|
||||
finalPayload = event.data as Record<string, unknown>;
|
||||
}
|
||||
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<string, unknown>;
|
||||
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<string, unknown>
|
||||
| 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<string, unknown> = {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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<Awaited<ReturnType<typeof getPersistedClawdTask>>>;
|
||||
}) {
|
||||
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) =>
|
||||
|
||||
@ -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变清晰处理失败",
|
||||
@ -1181,6 +1287,22 @@ export async function POST(request: NextRequest) {
|
||||
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) {
|
||||
if (fetchError.name === "AbortError") {
|
||||
@ -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 },
|
||||
|
||||
@ -32,6 +32,15 @@ const FAL_IMAGE_TASK_POLL_INTERVAL_MS = 2_000;
|
||||
const falTaskLastPolledAt = new Map<string, number>();
|
||||
const falTaskSyncInFlight = new Set<string>();
|
||||
|
||||
function toPublicVideoMetadata(value: DBMessage["videoMetadata"]) {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return value;
|
||||
const metadata = value as Record<string, unknown>;
|
||||
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,
|
||||
};
|
||||
});
|
||||
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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") && (
|
||||
<div className="space-y-2">
|
||||
<div className="flex items-center justify-between">
|
||||
|
||||
@ -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<string> {
|
||||
}
|
||||
}
|
||||
|
||||
async function normalizeDashscopeExpansionInput(imageUrl: string): Promise<string> {
|
||||
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(
|
||||
|
||||
@ -5,8 +5,16 @@ type ActiveClawdRequest = {
|
||||
expiresAt: number;
|
||||
};
|
||||
|
||||
const activeRequests = new Map<string, ActiveClawdRequest>();
|
||||
const canceledRequests = new Map<string, number>();
|
||||
const registry = globalThis as typeof globalThis & {
|
||||
__activeClawdRequests?: Map<string, ActiveClawdRequest>;
|
||||
__canceledClawdRequests?: Map<string, number>;
|
||||
};
|
||||
const activeRequests =
|
||||
registry.__activeClawdRequests || new Map<string, ActiveClawdRequest>();
|
||||
const canceledRequests =
|
||||
registry.__canceledClawdRequests || new Map<string, number>();
|
||||
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);
|
||||
|
||||
@ -20,11 +20,12 @@ export type AdminOpenClawChatConfig = Omit<OpenClawChatConfig, "token"> & {
|
||||
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<string, unknown> {
|
||||
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(
|
||||
|
||||
67
lib/openclaw-chat/task-secret.ts
Normal file
67
lib/openclaw-chat/task-secret.ts
Normal file
@ -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");
|
||||
}
|
||||
@ -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(),
|
||||
};
|
||||
|
||||
@ -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;
|
||||
|
||||
514
lib/openclaw-chat/upstream-task-client.ts
Normal file
514
lib/openclaw-chat/upstream-task-client.ts
Normal file
@ -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<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
}
|
||||
|
||||
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<string, unknown>;
|
||||
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<string, unknown> | 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<string, unknown>;
|
||||
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<OpenClawTaskOutcome> {
|
||||
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<void>((resolve, reject) => {
|
||||
if (signal?.aborted) {
|
||||
reject(signal.reason || new DOMException("Aborted", "AbortError"));
|
||||
return;
|
||||
}
|
||||
let timer: ReturnType<typeof setTimeout>;
|
||||
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();
|
||||
}
|
||||
@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
@ -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", () => {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user