fix(claw): resume durable long-running tasks
This commit is contained in:
parent
aebc7edecb
commit
7d67bd1059
@ -102,7 +102,10 @@ import {
|
|||||||
} from "@/components/canvas/canvas-skill-icons";
|
} from "@/components/canvas/canvas-skill-icons";
|
||||||
import { HomeAnimationEmbed } from "@/components/home/home-animation-embed";
|
import { HomeAnimationEmbed } from "@/components/home/home-animation-embed";
|
||||||
import {
|
import {
|
||||||
|
DEFAULT_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
||||||
|
MAX_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
|
MIN_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
OPENCLAW_CHAT_CONFIG_KEY,
|
OPENCLAW_CHAT_CONFIG_KEY,
|
||||||
type OpenClawChatConfig,
|
type OpenClawChatConfig,
|
||||||
} from "@/lib/openclaw-chat/config";
|
} from "@/lib/openclaw-chat/config";
|
||||||
@ -2241,7 +2244,7 @@ export function AdminSettingsClient() {
|
|||||||
enabled: false,
|
enabled: false,
|
||||||
baseUrl: DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
baseUrl: DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
||||||
token: "",
|
token: "",
|
||||||
timeoutMs: 600_000,
|
timeoutMs: DEFAULT_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
});
|
});
|
||||||
const [openClawChatTokenConfigured, setOpenClawChatTokenConfigured] =
|
const [openClawChatTokenConfigured, setOpenClawChatTokenConfigured] =
|
||||||
useState(false);
|
useState(false);
|
||||||
@ -2534,8 +2537,11 @@ export function AdminSettingsClient() {
|
|||||||
String(raw.baseUrl || "").trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
String(raw.baseUrl || "").trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
||||||
token: "",
|
token: "",
|
||||||
timeoutMs: Math.min(
|
timeoutMs: Math.min(
|
||||||
600_000,
|
MAX_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
Math.max(5_000, Math.round(Number(raw.timeoutMs) || 600_000))
|
Math.max(
|
||||||
|
MIN_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
|
Math.round(Number(raw.timeoutMs) || DEFAULT_OPENCLAW_TASK_TIMEOUT_MS)
|
||||||
|
)
|
||||||
),
|
),
|
||||||
});
|
});
|
||||||
setOpenClawChatTokenConfigured(
|
setOpenClawChatTokenConfigured(
|
||||||
@ -4443,8 +4449,14 @@ export function AdminSettingsClient() {
|
|||||||
openClawChatConfig.baseUrl.trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
openClawChatConfig.baseUrl.trim() || DEFAULT_OPENCLAW_CHAT_BASE_URL,
|
||||||
token: openClawChatConfig.token.trim(),
|
token: openClawChatConfig.token.trim(),
|
||||||
timeoutMs: Math.min(
|
timeoutMs: Math.min(
|
||||||
600_000,
|
MAX_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
Math.max(5_000, Math.round(Number(openClawChatConfig.timeoutMs) || 600_000))
|
Math.max(
|
||||||
|
MIN_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
|
Math.round(
|
||||||
|
Number(openClawChatConfig.timeoutMs) ||
|
||||||
|
DEFAULT_OPENCLAW_TASK_TIMEOUT_MS
|
||||||
|
)
|
||||||
|
)
|
||||||
),
|
),
|
||||||
};
|
};
|
||||||
|
|
||||||
@ -12817,21 +12829,27 @@ export function AdminSettingsClient() {
|
|||||||
</div>
|
</div>
|
||||||
<div className="space-y-2">
|
<div className="space-y-2">
|
||||||
<Label htmlFor="openClawChatTimeoutMs">
|
<Label htmlFor="openClawChatTimeoutMs">
|
||||||
请求超时(毫秒)
|
任务最长执行时间(毫秒)
|
||||||
</Label>
|
</Label>
|
||||||
<Input
|
<Input
|
||||||
id="openClawChatTimeoutMs"
|
id="openClawChatTimeoutMs"
|
||||||
min={5000}
|
min={MIN_OPENCLAW_TASK_TIMEOUT_MS}
|
||||||
max={600000}
|
max={MAX_OPENCLAW_TASK_TIMEOUT_MS}
|
||||||
onChange={(event) =>
|
onChange={(event) =>
|
||||||
setOpenClawChatConfig((prev) => ({
|
setOpenClawChatConfig((prev) => ({
|
||||||
...prev,
|
...prev,
|
||||||
timeoutMs: Number(event.target.value) || 600000,
|
timeoutMs:
|
||||||
|
Number(event.target.value) ||
|
||||||
|
DEFAULT_OPENCLAW_TASK_TIMEOUT_MS,
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
step={60_000}
|
||||||
type="number"
|
type="number"
|
||||||
value={openClawChatConfig.timeoutMs}
|
value={openClawChatConfig.timeoutMs}
|
||||||
/>
|
/>
|
||||||
|
<p className="text-muted-foreground text-xs">
|
||||||
|
网络断开会自动续接,该时间只限制业务任务的后台等待周期。
|
||||||
|
</p>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
<div className="space-y-2">
|
<div className="space-y-2">
|
||||||
|
|||||||
@ -1,7 +1,7 @@
|
|||||||
import { NextResponse } from "next/server";
|
import { NextResponse } from "next/server";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import { auth } from "@/app/(auth)/auth";
|
import { auth } from "@/app/(auth)/auth";
|
||||||
import { getChatById } from "@/lib/db/queries";
|
import { getChatById, getSystemSetting } from "@/lib/db/queries";
|
||||||
import {
|
import {
|
||||||
buildClawdCancelKey,
|
buildClawdCancelKey,
|
||||||
cancelClawdRequest,
|
cancelClawdRequest,
|
||||||
@ -10,7 +10,18 @@ import {
|
|||||||
OPENCLAW_AUTH_REQUIRED_CODE,
|
OPENCLAW_AUTH_REQUIRED_CODE,
|
||||||
OPENCLAW_AUTH_REQUIRED_MESSAGE,
|
OPENCLAW_AUTH_REQUIRED_MESSAGE,
|
||||||
} from "@/lib/openclaw-chat/errors";
|
} 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({
|
const cancelSchema = z.object({
|
||||||
id: z.string().uuid(),
|
id: z.string().uuid(),
|
||||||
@ -39,13 +50,34 @@ export async function POST(request: Request) {
|
|||||||
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
|
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
|
||||||
}
|
}
|
||||||
|
|
||||||
const active = cancelClawdRequest(
|
const cancelKey = buildClawdCancelKey({
|
||||||
buildClawdCancelKey({
|
userId: session.user.id,
|
||||||
userId: session.user.id,
|
chatId: parsed.data.id,
|
||||||
chatId: parsed.data.id,
|
userMessageId: parsed.data.userMessageId,
|
||||||
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({
|
const deleted = await deletePersistedClawdTask({
|
||||||
chatId: parsed.data.id,
|
chatId: parsed.data.id,
|
||||||
|
|||||||
@ -37,11 +37,17 @@ import {
|
|||||||
import {
|
import {
|
||||||
appendPersistedClawdTaskProgress,
|
appendPersistedClawdTaskProgress,
|
||||||
getPersistedClawdTask,
|
getPersistedClawdTask,
|
||||||
|
markClawdTaskCanceled,
|
||||||
markClawdTaskCompleted,
|
markClawdTaskCompleted,
|
||||||
markClawdTaskFailed,
|
markClawdTaskFailed,
|
||||||
markClawdTaskRunning,
|
markClawdTaskRunning,
|
||||||
persistPendingClawdTask,
|
persistPendingClawdTask,
|
||||||
} from "@/lib/openclaw-chat/task-store";
|
} from "@/lib/openclaw-chat/task-store";
|
||||||
|
import {
|
||||||
|
OpenClawTaskDetachedError,
|
||||||
|
submitOpenClawDurableTask,
|
||||||
|
waitForOpenClawDurableTask,
|
||||||
|
} from "@/lib/openclaw-chat/upstream-task-client";
|
||||||
import { generateUUID } from "@/lib/utils";
|
import { generateUUID } from "@/lib/utils";
|
||||||
|
|
||||||
const DEFAULT_PROCESSING_HINT =
|
const DEFAULT_PROCESSING_HINT =
|
||||||
@ -92,21 +98,6 @@ type ClawdSpreadsheetAttachment = {
|
|||||||
truncated: boolean;
|
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 = {
|
type ExecuteClawdTaskParams = {
|
||||||
assistantMessageId: string;
|
assistantMessageId: string;
|
||||||
cancelKey: string;
|
cancelKey: string;
|
||||||
@ -237,121 +228,6 @@ async function readOpenClawConfig() {
|
|||||||
return normalizeOpenClawChatConfig(setting?.value);
|
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) {
|
function getAnswer(payload: unknown) {
|
||||||
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
|
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
|
||||||
return "";
|
return "";
|
||||||
@ -367,145 +243,8 @@ function getRequestId(payload: unknown) {
|
|||||||
return value ? String(value) : null;
|
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) {
|
async function executeClawdTask(params: ExecuteClawdTaskParams) {
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
const timeout = setTimeout(() => controller.abort(), params.config.timeoutMs);
|
|
||||||
registerActiveClawdRequest(
|
registerActiveClawdRequest(
|
||||||
params.cancelKey,
|
params.cancelKey,
|
||||||
controller,
|
controller,
|
||||||
@ -519,8 +258,7 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
|
|||||||
assistantMessageId: params.assistantMessageId,
|
assistantMessageId: params.assistantMessageId,
|
||||||
});
|
});
|
||||||
|
|
||||||
const isSensitiveSubmission = Boolean(params.sensitiveInput);
|
const upstreamBody: Record<string, unknown> = {
|
||||||
const upstreamBody = JSON.stringify({
|
|
||||||
conversation_id: params.conversationId,
|
conversation_id: params.conversationId,
|
||||||
user_id: params.userId,
|
user_id: params.userId,
|
||||||
user_name: params.userName,
|
user_name: params.userName,
|
||||||
@ -560,64 +298,106 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
|
|||||||
params.spreadsheetMarkdownTruncated || undefined,
|
params.spreadsheetMarkdownTruncated || undefined,
|
||||||
...params.metadata,
|
...params.metadata,
|
||||||
},
|
},
|
||||||
stream_progress: !isSensitiveSubmission,
|
stream_progress: false,
|
||||||
sensitive_input: params.sensitiveInput,
|
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) {
|
if (params.sensitiveInput) {
|
||||||
throw new Error(`OpenClaw request failed: ${response.status}`);
|
const timeout = setTimeout(
|
||||||
}
|
() =>
|
||||||
|
controller.abort(
|
||||||
const contentType = response.headers.get("content-type") || "";
|
new DOMException("OpenClaw request timed out", "TimeoutError")
|
||||||
let payload = contentType.includes("text/event-stream")
|
),
|
||||||
? await readOpenClawSse({
|
params.config.timeoutMs
|
||||||
response,
|
);
|
||||||
chatId: params.chatId,
|
try {
|
||||||
assistantMessageId: params.assistantMessageId,
|
const response = await fetch(`${params.config.baseUrl}/api/chat`, {
|
||||||
})
|
method: "POST",
|
||||||
: ((await response.json().catch(() => null)) as Record<
|
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,
|
string,
|
||||||
unknown
|
unknown
|
||||||
> | null);
|
> | null;
|
||||||
if (!payload) throw new Error("OpenClaw returned an invalid response");
|
if (!response.ok) {
|
||||||
|
throw new Error(`OpenClaw request failed: ${response.status}`);
|
||||||
const backgroundTask = parseOpenClawBackgroundTask(payload);
|
}
|
||||||
if (backgroundTask?.state === "processing") {
|
if (!payload) throw new Error("OpenClaw returned an invalid response");
|
||||||
payload = await pollOpenClawBackgroundTask({
|
const answer = getAnswer(payload);
|
||||||
config: params.config,
|
if (!answer) throw new Error("OpenClaw returned an empty answer");
|
||||||
initialPayload: payload,
|
await markClawdTaskCompleted({
|
||||||
initialTask: backgroundTask,
|
chatId: params.chatId,
|
||||||
controller,
|
assistantMessageId: params.assistantMessageId,
|
||||||
chatId: params.chatId,
|
answer,
|
||||||
assistantMessageId: params.assistantMessageId,
|
requestId: getRequestId(payload),
|
||||||
userId: params.userId,
|
});
|
||||||
conversationId: params.conversationId,
|
clearClawdRequestCancellation(params.cancelKey);
|
||||||
});
|
return;
|
||||||
|
} finally {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (isClawdRequestCanceled(params.cancelKey)) return;
|
const target = {
|
||||||
const answer = getAnswer(payload);
|
|
||||||
if (!answer) throw new Error("OpenClaw returned an empty answer");
|
|
||||||
await markClawdTaskCompleted({
|
|
||||||
chatId: params.chatId,
|
chatId: params.chatId,
|
||||||
assistantMessageId: params.assistantMessageId,
|
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);
|
clearClawdRequestCancellation(params.cancelKey);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
if (isClawdRequestCanceled(params.cancelKey)) return;
|
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);
|
console.error("[ClawdChat API] Durable task failed:", error);
|
||||||
await markClawdTaskFailed({
|
await markClawdTaskFailed({
|
||||||
chatId: params.chatId,
|
chatId: params.chatId,
|
||||||
@ -632,7 +412,6 @@ async function executeClawdTask(params: ExecuteClawdTaskParams) {
|
|||||||
params.sensitiveInput.verification_code = "";
|
params.sensitiveInput.verification_code = "";
|
||||||
}
|
}
|
||||||
params.sensitiveInput = undefined;
|
params.sensitiveInput = undefined;
|
||||||
clearTimeout(timeout);
|
|
||||||
unregisterActiveClawdRequest(params.cancelKey);
|
unregisterActiveClawdRequest(params.cancelKey);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -1,12 +1,32 @@
|
|||||||
import { NextResponse } from "next/server";
|
import { NextResponse } from "next/server";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import { auth } from "@/app/(auth)/auth";
|
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 {
|
import {
|
||||||
OPENCLAW_AUTH_REQUIRED_CODE,
|
OPENCLAW_AUTH_REQUIRED_CODE,
|
||||||
OPENCLAW_AUTH_REQUIRED_MESSAGE,
|
OPENCLAW_AUTH_REQUIRED_MESSAGE,
|
||||||
} from "@/lib/openclaw-chat/errors";
|
} 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({
|
const statusSchema = z.object({
|
||||||
id: z.string().uuid(),
|
id: z.string().uuid(),
|
||||||
@ -15,6 +35,64 @@ const statusSchema = z.object({
|
|||||||
|
|
||||||
export const dynamic = "force-dynamic";
|
export const dynamic = "force-dynamic";
|
||||||
export const revalidate = 0;
|
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) {
|
export async function GET(request: Request) {
|
||||||
const session = await auth().catch(() => null);
|
const session = await auth().catch(() => null);
|
||||||
@ -45,7 +123,7 @@ export async function GET(request: Request) {
|
|||||||
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
|
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
|
||||||
}
|
}
|
||||||
|
|
||||||
const task = await getPersistedClawdTask({
|
let task = await getPersistedClawdTask({
|
||||||
chatId: parsed.data.id,
|
chatId: parsed.data.id,
|
||||||
userMessageId: parsed.data.userMessageId,
|
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)
|
const userTextPart = Array.isArray(task.userMessage?.parts)
|
||||||
? task.userMessage.parts.find(
|
? task.userMessage.parts.find(
|
||||||
(part) =>
|
(part) =>
|
||||||
|
|||||||
@ -32,6 +32,15 @@ const FAL_IMAGE_TASK_POLL_INTERVAL_MS = 2_000;
|
|||||||
const falTaskLastPolledAt = new Map<string, number>();
|
const falTaskLastPolledAt = new Map<string, number>();
|
||||||
const falTaskSyncInFlight = new Set<string>();
|
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() {
|
function createEmptyMessagesResponse() {
|
||||||
return NextResponse.json(
|
return NextResponse.json(
|
||||||
{ messages: [] },
|
{ messages: [] },
|
||||||
@ -277,7 +286,7 @@ export async function GET(request: NextRequest) {
|
|||||||
createdAt: msg.createdAt,
|
createdAt: msg.createdAt,
|
||||||
parts: parts,
|
parts: parts,
|
||||||
attachments: msg.attachments || [],
|
attachments: msg.attachments || [],
|
||||||
videoMetadata: msg.videoMetadata || null,
|
videoMetadata: toPublicVideoMetadata(msg.videoMetadata) || null,
|
||||||
};
|
};
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@ -5,8 +5,16 @@ type ActiveClawdRequest = {
|
|||||||
expiresAt: number;
|
expiresAt: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
const activeRequests = new Map<string, ActiveClawdRequest>();
|
const registry = globalThis as typeof globalThis & {
|
||||||
const canceledRequests = new Map<string, number>();
|
__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()) {
|
function cleanupExpiredClawdRequests(now = Date.now()) {
|
||||||
for (const [key, request] of activeRequests.entries()) {
|
for (const [key, request] of activeRequests.entries()) {
|
||||||
@ -50,6 +58,11 @@ export function unregisterActiveClawdRequest(key: string) {
|
|||||||
activeRequests.delete(key);
|
activeRequests.delete(key);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function hasActiveClawdRequest(key: string) {
|
||||||
|
cleanupExpiredClawdRequests();
|
||||||
|
return activeRequests.has(key);
|
||||||
|
}
|
||||||
|
|
||||||
export function cancelClawdRequest(key: string) {
|
export function cancelClawdRequest(key: string) {
|
||||||
cleanupExpiredClawdRequests();
|
cleanupExpiredClawdRequests();
|
||||||
canceledRequests.set(key, Date.now() + CANCELED_REQUEST_TTL_MS);
|
canceledRequests.set(key, Date.now() + CANCELED_REQUEST_TTL_MS);
|
||||||
|
|||||||
@ -20,11 +20,12 @@ export type AdminOpenClawChatConfig = Omit<OpenClawChatConfig, "token"> & {
|
|||||||
tokenPreview: string;
|
tokenPreview: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
const DEFAULT_TIMEOUT_MS = 30 * 60 * 1000;
|
export const DEFAULT_OPENCLAW_TASK_TIMEOUT_MS = 30 * 60 * 1000;
|
||||||
const MIN_TIMEOUT_MS = 5_000;
|
export const MIN_OPENCLAW_TASK_TIMEOUT_MS = 5 * 60 * 1000;
|
||||||
// Complex OpenClaw skills can legitimately run for up to thirty minutes.
|
// Complex OpenClaw skills can legitimately run for up to thirty minutes.
|
||||||
// Keep this aligned with the route maxDuration and the integration contract.
|
// 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> {
|
function readRecord(value: unknown): Record<string, unknown> {
|
||||||
return value && typeof value === "object" && !Array.isArray(value)
|
return value && typeof value === "object" && !Array.isArray(value)
|
||||||
@ -41,8 +42,16 @@ function normalizeBaseUrl(value: unknown) {
|
|||||||
|
|
||||||
function normalizeTimeoutMs(value: unknown) {
|
function normalizeTimeoutMs(value: unknown) {
|
||||||
const numeric = Number(value);
|
const numeric = Number(value);
|
||||||
if (!Number.isFinite(numeric)) return DEFAULT_TIMEOUT_MS;
|
if (!Number.isFinite(numeric)) return DEFAULT_OPENCLAW_TASK_TIMEOUT_MS;
|
||||||
return Math.min(MAX_TIMEOUT_MS, Math.max(MIN_TIMEOUT_MS, Math.round(numeric)));
|
// 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(
|
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;
|
skillTagName?: string | null;
|
||||||
requestId?: string | null;
|
requestId?: string | null;
|
||||||
error?: string | null;
|
error?: string | null;
|
||||||
|
upstreamTaskId?: string | null;
|
||||||
|
upstreamTaskKey?: string | null;
|
||||||
|
upstreamEventsEndpoint?: string | null;
|
||||||
|
upstreamLastEventId?: number | null;
|
||||||
createdAt: number;
|
createdAt: number;
|
||||||
updatedAt: number;
|
updatedAt: number;
|
||||||
};
|
};
|
||||||
@ -56,6 +60,7 @@ export function normalizeClawdTaskMetadata(
|
|||||||
|
|
||||||
const createdAt = Number(raw.createdAt);
|
const createdAt = Number(raw.createdAt);
|
||||||
const updatedAt = Number(raw.updatedAt);
|
const updatedAt = Number(raw.updatedAt);
|
||||||
|
const upstreamLastEventId = Number(raw.upstreamLastEventId);
|
||||||
return {
|
return {
|
||||||
composerMode: "clawd",
|
composerMode: "clawd",
|
||||||
source: "openclaw",
|
source: "openclaw",
|
||||||
@ -74,6 +79,14 @@ export function normalizeClawdTaskMetadata(
|
|||||||
skillTagName: raw.skillTagName ? String(raw.skillTagName) : null,
|
skillTagName: raw.skillTagName ? String(raw.skillTagName) : null,
|
||||||
requestId: raw.requestId ? String(raw.requestId) : null,
|
requestId: raw.requestId ? String(raw.requestId) : null,
|
||||||
error: raw.error ? String(raw.error) : 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(),
|
createdAt: Number.isFinite(createdAt) ? createdAt : Date.now(),
|
||||||
updatedAt: Number.isFinite(updatedAt) ? updatedAt : Date.now(),
|
updatedAt: Number.isFinite(updatedAt) ? updatedAt : Date.now(),
|
||||||
};
|
};
|
||||||
|
|||||||
@ -8,6 +8,7 @@ import {
|
|||||||
updateMessageById,
|
updateMessageById,
|
||||||
} from "@/lib/db/queries";
|
} from "@/lib/db/queries";
|
||||||
import type { DBMessage } from "@/lib/db/schema";
|
import type { DBMessage } from "@/lib/db/schema";
|
||||||
|
import { encryptOpenClawTaskSecret } from "@/lib/openclaw-chat/task-secret";
|
||||||
import {
|
import {
|
||||||
appendClawdTaskProgress,
|
appendClawdTaskProgress,
|
||||||
CLAWD_TASK_KIND,
|
CLAWD_TASK_KIND,
|
||||||
@ -29,6 +30,13 @@ type PendingClawdTaskParams = {
|
|||||||
processingHint: string;
|
processingHint: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export type ClawdUpstreamTaskState = {
|
||||||
|
taskId: string;
|
||||||
|
taskKey?: string | null;
|
||||||
|
eventsEndpoint?: string | null;
|
||||||
|
lastEventId?: number | null;
|
||||||
|
};
|
||||||
|
|
||||||
function buildTaskMetadata(params: {
|
function buildTaskMetadata(params: {
|
||||||
userMessageId: string;
|
userMessageId: string;
|
||||||
assistantMessageId: string;
|
assistantMessageId: string;
|
||||||
@ -39,6 +47,7 @@ function buildTaskMetadata(params: {
|
|||||||
progressEvents?: ClawdTaskProgressPayload[];
|
progressEvents?: ClawdTaskProgressPayload[];
|
||||||
requestId?: string | null;
|
requestId?: string | null;
|
||||||
error?: string | null;
|
error?: string | null;
|
||||||
|
upstreamTask?: ClawdUpstreamTaskState | null;
|
||||||
createdAt?: number;
|
createdAt?: number;
|
||||||
updatedAt?: number;
|
updatedAt?: number;
|
||||||
}): ClawdTaskMetadata {
|
}): ClawdTaskMetadata {
|
||||||
@ -56,6 +65,12 @@ function buildTaskMetadata(params: {
|
|||||||
skillTagName: params.skillTagName || null,
|
skillTagName: params.skillTagName || null,
|
||||||
requestId: params.requestId || null,
|
requestId: params.requestId || null,
|
||||||
error: params.error || 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,
|
createdAt: params.createdAt || now,
|
||||||
updatedAt: params.updatedAt || now,
|
updatedAt: params.updatedAt || now,
|
||||||
};
|
};
|
||||||
@ -134,6 +149,7 @@ async function updateClawdAssistantTask(params: {
|
|||||||
parts?: DBMessage["parts"];
|
parts?: DBMessage["parts"];
|
||||||
requestId?: string | null;
|
requestId?: string | null;
|
||||||
error?: string | null;
|
error?: string | null;
|
||||||
|
upstreamTask?: ClawdUpstreamTaskState;
|
||||||
}) {
|
}) {
|
||||||
const rows = await getMessageById({ id: params.assistantMessageId });
|
const rows = await getMessageById({ id: params.assistantMessageId });
|
||||||
const message = rows.find((row) => row.chatId === params.chatId);
|
const message = rows.find((row) => row.chatId === params.chatId);
|
||||||
@ -151,6 +167,22 @@ async function updateClawdAssistantTask(params: {
|
|||||||
requestId:
|
requestId:
|
||||||
params.requestId !== undefined ? params.requestId : metadata.requestId,
|
params.requestId !== undefined ? params.requestId : metadata.requestId,
|
||||||
error: params.error !== undefined ? params.error : metadata.error,
|
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(),
|
updatedAt: Date.now(),
|
||||||
};
|
};
|
||||||
await updateMessageById({
|
await updateMessageById({
|
||||||
@ -173,11 +205,26 @@ export async function appendPersistedClawdTaskProgress(params: {
|
|||||||
chatId: string;
|
chatId: string;
|
||||||
assistantMessageId: string;
|
assistantMessageId: string;
|
||||||
progress: ClawdTaskProgressPayload;
|
progress: ClawdTaskProgressPayload;
|
||||||
|
upstreamTask?: ClawdUpstreamTaskState;
|
||||||
}) {
|
}) {
|
||||||
return updateClawdAssistantTask({
|
return updateClawdAssistantTask({
|
||||||
...params,
|
...params,
|
||||||
status: "running",
|
status: "running",
|
||||||
progress: params.progress,
|
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: {
|
export async function getPersistedClawdTask(params: {
|
||||||
chatId: string;
|
chatId: string;
|
||||||
userMessageId?: string | null;
|
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
|
}).timeoutMs
|
||||||
).toBe(30 * 60 * 1000);
|
).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",
|
assistantMessageId: "assistant-message-id",
|
||||||
processingHint: "正在处理",
|
processingHint: "正在处理",
|
||||||
progressEvents: [{ event_id: 1, message: "请求已接收" }],
|
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,
|
createdAt: 100,
|
||||||
updatedAt: 200,
|
updatedAt: 200,
|
||||||
});
|
});
|
||||||
@ -26,6 +30,8 @@ test.describe("OpenClaw durable task state", () => {
|
|||||||
expect(task?.progressEvents).toEqual([
|
expect(task?.progressEvents).toEqual([
|
||||||
{ event_id: 1, message: "请求已接收" },
|
{ 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", () => {
|
test("deduplicates replayed progress events while keeping new steps", () => {
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user