fix: stream Claw background task progress
This commit is contained in:
134
parent
000e5c75c9
commit
55be81d3d8
@ -232,6 +232,126 @@ function parseSseBlock(block: string) {
|
||||
}
|
||||
}
|
||||
|
||||
type OpenClawBackgroundTask = {
|
||||
jobId: string;
|
||||
state: string;
|
||||
stage: string;
|
||||
phase: string;
|
||||
displayText: string;
|
||||
updatedAt: string;
|
||||
completed: number;
|
||||
total: number;
|
||||
progressPercent: number;
|
||||
pollEndpoint: string;
|
||||
pollAfterMs: number;
|
||||
maxWaitSeconds: number;
|
||||
autoNotify: boolean;
|
||||
};
|
||||
|
||||
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();
|
||||
const pollEndpoint = String(rawTask.poll_endpoint || "").trim();
|
||||
if (!jobId || !state) return null;
|
||||
|
||||
const completed = Math.max(0, Number(rawTask.completed) || 0);
|
||||
const total = Math.max(1, Number(rawTask.total) || 1);
|
||||
const progressPercent = Math.max(
|
||||
0,
|
||||
Math.min(
|
||||
100,
|
||||
Number.isFinite(Number(rawTask.progress_percent))
|
||||
? Number(rawTask.progress_percent)
|
||||
: Math.round((completed * 100) / total)
|
||||
)
|
||||
);
|
||||
|
||||
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,
|
||||
pollEndpoint,
|
||||
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)
|
||||
),
|
||||
autoNotify: rawTask.auto_notify === true,
|
||||
};
|
||||
}
|
||||
|
||||
function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) {
|
||||
return {
|
||||
step_code: task.stage || "background_task",
|
||||
step_state:
|
||||
task.state === "failed"
|
||||
? "failed"
|
||||
: task.state === "processing"
|
||||
? "started"
|
||||
: "completed",
|
||||
message:
|
||||
task.displayText ||
|
||||
task.phase ||
|
||||
(task.state === "processing" ? "任务正在处理" : "任务状态已更新"),
|
||||
display:
|
||||
task.displayText ||
|
||||
task.phase ||
|
||||
(task.state === "processing" ? "任务正在处理" : "任务状态已更新"),
|
||||
completed: task.completed,
|
||||
total: task.total,
|
||||
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;
|
||||
}
|
||||
|
||||
async function persistClawdResponse({
|
||||
assistantMessageId,
|
||||
attachments,
|
||||
@ -361,6 +481,181 @@ async function persistClawdResponse({
|
||||
};
|
||||
}
|
||||
|
||||
function streamOpenClawBackgroundTask(params: {
|
||||
request: Request;
|
||||
cancelKey: string;
|
||||
config: { baseUrl: string; token: string; timeoutMs: number };
|
||||
initialPayload: Record<string, unknown>;
|
||||
initialTask: OpenClawBackgroundTask;
|
||||
assistantMessageId: string;
|
||||
attachments: OpenClawChatMediaAttachment[];
|
||||
chatId: string;
|
||||
conversationId: string;
|
||||
displayMessage: string;
|
||||
skillTagId: string;
|
||||
skillTagName: string;
|
||||
spreadsheetAttachments: ClawdSpreadsheetAttachment[];
|
||||
userId: string;
|
||||
userMessageId: string;
|
||||
}) {
|
||||
const encoder = new TextEncoder();
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
async start(streamController) {
|
||||
const controller = new AbortController();
|
||||
const handleRequestAbort = () => controller.abort();
|
||||
params.request.signal.addEventListener("abort", handleRequestAbort, {
|
||||
once: true,
|
||||
});
|
||||
registerActiveClawdRequest(
|
||||
params.cancelKey,
|
||||
controller,
|
||||
Math.max(params.config.timeoutMs, 60 * 60 * 1000)
|
||||
);
|
||||
|
||||
let task = params.initialTask;
|
||||
let finalPayload = params.initialPayload;
|
||||
let lastProgressKey = "";
|
||||
let consecutiveFailures = 0;
|
||||
|
||||
const emitProgress = (nextTask: OpenClawBackgroundTask) => {
|
||||
const progress = buildOpenClawTaskProgress(nextTask);
|
||||
const key = [
|
||||
nextTask.state,
|
||||
nextTask.stage,
|
||||
nextTask.updatedAt,
|
||||
nextTask.completed,
|
||||
nextTask.total,
|
||||
progress.message,
|
||||
].join("|");
|
||||
if (key === lastProgressKey) return;
|
||||
lastProgressKey = key;
|
||||
streamController.enqueue(
|
||||
encoder.encode(encodeSseEvent("progress", progress))
|
||||
);
|
||||
};
|
||||
|
||||
try {
|
||||
emitProgress(task);
|
||||
while (task.state === "processing" && task.autoNotify) {
|
||||
if (
|
||||
controller.signal.aborted ||
|
||||
isClawdRequestCanceled(params.cancelKey)
|
||||
) {
|
||||
throw new DOMException("Claw request canceled", "AbortError");
|
||||
}
|
||||
|
||||
const statusUrl = buildOpenClawTaskStatusUrl({
|
||||
baseUrl: params.config.baseUrl,
|
||||
task,
|
||||
userId: params.userId,
|
||||
conversationId: params.conversationId,
|
||||
});
|
||||
let statusResponse: Response;
|
||||
try {
|
||||
statusResponse = await fetch(statusUrl, {
|
||||
headers: {
|
||||
Authorization: `Bearer ${params.config.token}`,
|
||||
Accept: "application/json",
|
||||
},
|
||||
cache: "no-store",
|
||||
signal: controller.signal,
|
||||
});
|
||||
} catch (error) {
|
||||
if (controller.signal.aborted) throw error;
|
||||
consecutiveFailures += 1;
|
||||
if (consecutiveFailures >= 5) throw error;
|
||||
streamController.enqueue(
|
||||
encoder.encode(
|
||||
encodeSseEvent("progress", {
|
||||
step_code: task.stage || "background_task",
|
||||
step_state: "waiting",
|
||||
message: "任务仍在运行，正在重新连接进度服务",
|
||||
display: "任务仍在运行，正在重新连接进度服务",
|
||||
completed: task.completed,
|
||||
total: task.total,
|
||||
progress_percent: task.progressPercent,
|
||||
})
|
||||
)
|
||||
);
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, task.pollAfterMs)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!statusResponse.ok) {
|
||||
throw new Error(`OpenClaw task status failed: ${statusResponse.status}`);
|
||||
}
|
||||
const statusPayload = (await statusResponse.json().catch(() => null)) as
|
||||
| Record<string, unknown>
|
||||
| null;
|
||||
const nextTask = parseOpenClawBackgroundTask(statusPayload);
|
||||
if (!statusPayload || !nextTask) {
|
||||
throw new Error("OpenClaw task status payload is invalid");
|
||||
}
|
||||
|
||||
consecutiveFailures = 0;
|
||||
finalPayload = {
|
||||
...statusPayload,
|
||||
request_id:
|
||||
statusPayload.request_id || params.initialPayload.request_id || null,
|
||||
};
|
||||
task = nextTask;
|
||||
emitProgress(task);
|
||||
}
|
||||
|
||||
const persisted = await persistClawdResponse({
|
||||
assistantMessageId: params.assistantMessageId,
|
||||
attachments: params.attachments,
|
||||
chatId: params.chatId,
|
||||
conversationId: params.conversationId,
|
||||
displayMessage: params.displayMessage,
|
||||
skillTagId: params.skillTagId,
|
||||
skillTagName: params.skillTagName,
|
||||
spreadsheetAttachments: params.spreadsheetAttachments,
|
||||
upstreamPayload: finalPayload,
|
||||
userMessageId: params.userMessageId,
|
||||
});
|
||||
clearClawdRequestCancellation(params.cancelKey);
|
||||
streamController.enqueue(
|
||||
encoder.encode(encodeSseEvent("final", persisted))
|
||||
);
|
||||
} catch (error) {
|
||||
const wasCanceled =
|
||||
controller.signal.aborted ||
|
||||
isClawdRequestCanceled(params.cancelKey) ||
|
||||
(error instanceof Error && error.name === "AbortError");
|
||||
streamController.enqueue(
|
||||
encoder.encode(
|
||||
encodeSseEvent(
|
||||
wasCanceled ? "final" : "error",
|
||||
wasCanceled
|
||||
? { canceled: true }
|
||||
: { error: CLAW_UPGRADE_RETRY_MESSAGE }
|
||||
)
|
||||
)
|
||||
);
|
||||
} finally {
|
||||
params.request.signal.removeEventListener("abort", handleRequestAbort);
|
||||
unregisterActiveClawdRequest(params.cancelKey);
|
||||
try {
|
||||
streamController.close();
|
||||
} catch {
|
||||
// The browser may already have closed the stream.
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
return new Response(stream, {
|
||||
headers: {
|
||||
"Cache-Control": "no-cache, no-store",
|
||||
"Content-Type": "text/event-stream; charset=utf-8",
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function POST(request: Request) {
|
||||
const session = await auth().catch(() => null);
|
||||
|
||||
@ -500,6 +795,7 @@ export async function POST(request: Request) {
|
||||
headers: {
|
||||
Authorization: `Bearer ${config.token}`,
|
||||
"Content-Type": "application/json",
|
||||
Accept: "text/event-stream",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
conversation_id: conversationId,
|
||||
@ -657,6 +953,31 @@ export async function POST(request: Request) {
|
||||
{ status: response.status }
|
||||
);
|
||||
}
|
||||
|
||||
const backgroundTask = parseOpenClawBackgroundTask(upstreamPayload);
|
||||
if (
|
||||
backgroundTask?.state === "processing" &&
|
||||
backgroundTask.autoNotify &&
|
||||
backgroundTask.pollEndpoint
|
||||
) {
|
||||
return streamOpenClawBackgroundTask({
|
||||
request,
|
||||
cancelKey,
|
||||
config,
|
||||
initialPayload: upstreamPayload as Record<string, unknown>,
|
||||
initialTask: backgroundTask,
|
||||
assistantMessageId,
|
||||
attachments: parsed.attachments,
|
||||
chatId: parsed.id,
|
||||
conversationId,
|
||||
displayMessage,
|
||||
skillTagId,
|
||||
skillTagName,
|
||||
spreadsheetAttachments,
|
||||
userId: session.user.id,
|
||||
userMessageId,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
unregisterActiveClawdRequest(cancelKey);
|
||||
const errorName =
|
||||
|
||||
@ -494,6 +494,65 @@ type PendingClawdRequestRestoreSnapshot = {
|
||||
shouldRestoreSkillTag: boolean;
|
||||
};
|
||||
|
||||
type ClawdProgressEvent = {
|
||||
key: string;
|
||||
eventId?: number;
|
||||
sequence?: number;
|
||||
stepCode: string;
|
||||
stepState: string;
|
||||
message: string;
|
||||
currentOrderId: string;
|
||||
completed?: number;
|
||||
total?: number;
|
||||
progressPercent?: number;
|
||||
};
|
||||
|
||||
function normalizeClawdProgressEvent(
|
||||
progress: Record<string, unknown>
|
||||
): ClawdProgressEvent | null {
|
||||
if (progress.heartbeat === true) return null;
|
||||
const message = String(
|
||||
progress.message || progress.display || ""
|
||||
).trim();
|
||||
if (!message) return null;
|
||||
const eventId = Number(progress.event_id);
|
||||
const sequence = Number(progress.sequence);
|
||||
const stepCode = String(progress.step_code || "").trim();
|
||||
const stepState = String(progress.step_state || "").trim();
|
||||
const currentOrderId = String(progress.current_order_id || "").trim();
|
||||
const completed = Number(progress.completed);
|
||||
const total = Number(progress.total);
|
||||
const progressPercent = Number(progress.progress_percent);
|
||||
const normalizedEventId = Number.isFinite(eventId) && eventId > 0
|
||||
? eventId
|
||||
: undefined;
|
||||
const normalizedSequence = Number.isFinite(sequence) && sequence > 0
|
||||
? sequence
|
||||
: undefined;
|
||||
|
||||
return {
|
||||
key:
|
||||
normalizedEventId != null
|
||||
? `event-${normalizedEventId}`
|
||||
: `sequence-${normalizedSequence || 0}-${stepCode}-${message}`,
|
||||
eventId: normalizedEventId,
|
||||
sequence: normalizedSequence,
|
||||
stepCode,
|
||||
stepState,
|
||||
message,
|
||||
currentOrderId,
|
||||
completed:
|
||||
Number.isFinite(completed) && completed >= 0 ? completed : undefined,
|
||||
total: Number.isFinite(total) && total > 0 ? total : undefined,
|
||||
progressPercent:
|
||||
Number.isFinite(progressPercent) &&
|
||||
progressPercent >= 0 &&
|
||||
progressPercent <= 100
|
||||
? Math.round(progressPercent)
|
||||
: undefined,
|
||||
};
|
||||
}
|
||||
|
||||
function isClawdSubmittedComposerText(
|
||||
snapshot: ClawdComposerSubmissionSnapshot,
|
||||
value: string
|
||||
@ -1924,16 +1983,85 @@ function CanvasAssistantTypingPlaceholder({
|
||||
label = "分析中",
|
||||
onCancel,
|
||||
cancelLabel = "停止",
|
||||
progressEvents = [],
|
||||
}: {
|
||||
hint?: string | null;
|
||||
label?: string;
|
||||
onCancel?: () => void;
|
||||
cancelLabel?: string;
|
||||
progressEvents?: ClawdProgressEvent[];
|
||||
}) {
|
||||
const visibleProgressEvents = progressEvents.slice(-12);
|
||||
const hiddenProgressCount = Math.max(
|
||||
0,
|
||||
progressEvents.length - visibleProgressEvents.length
|
||||
);
|
||||
return (
|
||||
<div className="flex w-full justify-start">
|
||||
<div className="flex max-w-[min(78%,680px)] flex-col items-start gap-2">
|
||||
{hint ? (
|
||||
<div className="flex w-full max-w-[min(92%,680px)] flex-col items-start gap-2 sm:max-w-[min(78%,680px)]">
|
||||
{visibleProgressEvents.length > 0 ? (
|
||||
<div
|
||||
className="w-full min-w-0 rounded-lg bg-[#f4f5f7] px-4 py-3 text-[#2f343d] shadow-sm dark:bg-white/8 dark:text-zinc-200"
|
||||
aria-live="polite"
|
||||
>
|
||||
<div className="mb-2 flex items-center justify-between gap-3 text-[12px] font-medium text-[#737d8c] dark:text-zinc-400">
|
||||
<span>执行进度</span>
|
||||
<span>{progressEvents.length} 步</span>
|
||||
</div>
|
||||
{hiddenProgressCount > 0 ? (
|
||||
<div className="mb-1.5 pl-6 text-[11px] text-[#9aa3b2] dark:text-zinc-500">
|
||||
已记录此前 {hiddenProgressCount} 步
|
||||
</div>
|
||||
) : null}
|
||||
<div className="max-h-64 space-y-1.5 overflow-y-auto pr-1">
|
||||
{visibleProgressEvents.map((progressEvent, index) => {
|
||||
const isLatest = index === visibleProgressEvents.length - 1;
|
||||
const isFailed = progressEvent.stepState === "failed";
|
||||
const isActive =
|
||||
isLatest &&
|
||||
["accepted", "queued", "started", "waiting"].includes(
|
||||
progressEvent.stepState
|
||||
);
|
||||
return (
|
||||
<div
|
||||
key={progressEvent.key}
|
||||
className="flex min-w-0 items-start gap-2 text-[13px] leading-5"
|
||||
>
|
||||
{isFailed ? (
|
||||
<AlertCircle className="mt-0.5 h-4 w-4 shrink-0 text-rose-500" />
|
||||
) : isActive ? (
|
||||
<Loader2 className="mt-0.5 h-4 w-4 shrink-0 animate-spin text-[#617086] dark:text-zinc-300" />
|
||||
) : (
|
||||
<CheckCircle2 className="mt-0.5 h-4 w-4 shrink-0 text-emerald-600 dark:text-emerald-400" />
|
||||
)}
|
||||
<div className="min-w-0">
|
||||
<div className="break-words">{progressEvent.message}</div>
|
||||
{progressEvent.currentOrderId ||
|
||||
progressEvent.total ? (
|
||||
<div className="text-[11px] text-[#929baa] dark:text-zinc-500">
|
||||
{progressEvent.currentOrderId
|
||||
? `订单 ${progressEvent.currentOrderId}`
|
||||
: ""}
|
||||
{progressEvent.currentOrderId &&
|
||||
progressEvent.total
|
||||
? " · "
|
||||
: ""}
|
||||
{progressEvent.total
|
||||
? `${progressEvent.completed || 0}/${progressEvent.total}${
|
||||
progressEvent.progressPercent != null
|
||||
? ` · ${progressEvent.progressPercent}%`
|
||||
: ""
|
||||
}`
|
||||
: ""}
|
||||
</div>
|
||||
) : null}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
</div>
|
||||
) : hint ? (
|
||||
<div className="rounded-[18px] bg-[#f4f5f7] px-4 py-3 text-[15px] leading-6 text-[#2f343d] shadow-sm dark:bg-white/8 dark:text-zinc-200">
|
||||
{hint}
|
||||
</div>
|
||||
@ -2493,6 +2621,9 @@ export function CanvasChatPanel({
|
||||
const [isDetailGenerating, setIsDetailGenerating] = useState(false);
|
||||
const [isClawdSubmitting, setIsClawdSubmitting] = useState(false);
|
||||
const [clawdProcessingHint, setClawdProcessingHint] = useState<string | null>(null);
|
||||
const [clawdProgressEvents, setClawdProgressEvents] = useState<
|
||||
ClawdProgressEvent[]
|
||||
>([]);
|
||||
const [detailSections, setDetailSections] = useState<Record<DetailSectionKey, boolean>>({
|
||||
required: true,
|
||||
strategy: false,
|
||||
@ -7804,6 +7935,7 @@ export function CanvasChatPanel({
|
||||
clawdAbortControllerRef.current = null;
|
||||
setIsClawdSubmitting(false);
|
||||
setClawdProcessingHint(null);
|
||||
setClawdProgressEvents([]);
|
||||
releaseSendGuard(snapshot.chatId);
|
||||
restorePendingClawdRequestToComposer(snapshot);
|
||||
toast.message("已停止处理，可以重新编辑后再发送");
|
||||
@ -11527,6 +11659,7 @@ export function CanvasChatPanel({
|
||||
clawdAbortControllerRef.current = clawdAbortController;
|
||||
pendingClawdRequestRestoreRef.current = clawdRequestRestoreSnapshot;
|
||||
setClawdProcessingHint(nextProcessingHint);
|
||||
setClawdProgressEvents([]);
|
||||
setIsClawdSubmitting(true);
|
||||
setMessages((previousMessages) =>
|
||||
sortCanvasMessagesByCreatedAt([
|
||||
@ -11584,6 +11717,19 @@ export function CanvasChatPanel({
|
||||
if (display) {
|
||||
setClawdProcessingHint(display);
|
||||
}
|
||||
const normalizedProgress = normalizeClawdProgressEvent(progress);
|
||||
if (normalizedProgress) {
|
||||
setClawdProgressEvents((previousEvents) => {
|
||||
if (
|
||||
previousEvents.some(
|
||||
(item) => item.key === normalizedProgress.key
|
||||
)
|
||||
) {
|
||||
return previousEvents;
|
||||
}
|
||||
return [...previousEvents, normalizedProgress].slice(-2000);
|
||||
});
|
||||
}
|
||||
});
|
||||
if (response.status === 499 || responseData?.canceled === true) {
|
||||
canceledClawdMessageIdsRef.current.add(userMessageId);
|
||||
@ -11713,6 +11859,7 @@ export function CanvasChatPanel({
|
||||
}
|
||||
setIsClawdSubmitting(false);
|
||||
setClawdProcessingHint(null);
|
||||
setClawdProgressEvents([]);
|
||||
releaseSendGuard(activeChatId);
|
||||
}
|
||||
return;
|
||||
@ -15183,6 +15330,7 @@ export function CanvasChatPanel({
|
||||
hint={clawdProcessingHint}
|
||||
label="处理中"
|
||||
onCancel={handleCancelClawdSubmission}
|
||||
progressEvents={clawdProgressEvents}
|
||||
/>
|
||||
) : null}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user