547 lines
16 KiB
TypeScript
547 lines
16 KiB
TypeScript
import "server-only";
|
|
|
|
import {
|
|
appendPersistedClawdTaskProgress,
|
|
findPersistedClawdUpstreamTaskState,
|
|
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 responseTaskKey = readText(root.task_key);
|
|
const persistedTask = responseTaskKey
|
|
? null
|
|
: await findPersistedClawdUpstreamTaskState({
|
|
chatId: params.target.chatId,
|
|
taskId,
|
|
});
|
|
const taskKey = responseTaskKey || persistedTask?.taskKey || null;
|
|
if (!taskKey) {
|
|
throw new OpenClawTaskCredentialUnavailableError(taskId);
|
|
}
|
|
const submittedTask = mergeUpstreamTaskState(
|
|
{
|
|
taskId,
|
|
taskKey,
|
|
eventsEndpoint:
|
|
getTaskEventsEndpoint(payload, taskId) ||
|
|
persistedTask?.eventsEndpoint ||
|
|
null,
|
|
lastEventId: persistedTask?.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,
|
|
taskKey,
|
|
lastEventId: persistedTask?.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> {
|
|
if (!params.upstreamTask.taskKey) {
|
|
throw new OpenClawTaskCredentialUnavailableError(
|
|
params.upstreamTask.taskId
|
|
);
|
|
}
|
|
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 class OpenClawTaskCredentialUnavailableError extends OpenClawTaskDetachedError {
|
|
constructor(readonly taskId: string) {
|
|
super();
|
|
this.name = "OpenClawTaskCredentialUnavailableError";
|
|
this.message = "OpenClaw task key is unavailable; keep the task detached";
|
|
}
|
|
}
|
|
|
|
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();
|
|
}
|