diff --git a/app/(chat)/api/chat/route.ts b/app/(chat)/api/chat/route.ts index e1971a923..e95340dc8 100644 --- a/app/(chat)/api/chat/route.ts +++ b/app/(chat)/api/chat/route.ts @@ -1,10 +1,11 @@ import { isExplicitDestructiveIntent } from "@/lib/ai/canvas-agent-runtime/capability-access"; import { inferVideoTaskRequirements } from "@/lib/ai/canvas-agent-runtime/video-protocol"; import { createLogger } from "@/lib/observability/logger"; +import { trackRuntimeWork } from "@/lib/observability/active-work"; import { geolocation } from "@vercel/functions"; import { convertToModelMessages, - createUIMessageStream, + createUIMessageStream as createAiUIMessageStream, generateText, smoothStream, stepCountIs, @@ -426,6 +427,8 @@ import { } from "./_lib/video-models"; const chatApiLogger = createLogger("chat.route"); +const createUIMessageStream: typeof createAiUIMessageStream = (options) => + createAiUIMessageStream({ ...options, execute: (writer) => trackRuntimeWork(() => options.execute(writer)) }); const INTERNAL_IMAGE_GENERATION_FAILURE_PATTERN = /生成图片失败|图片生成失败|未找到图片数据|未解析到图片|模型未返回生成图片|图片数据不可用|未返回可用图片/i; @@ -443,13 +446,13 @@ function assertGeneratedImageMarkdownHasImage(markdown: string, modelLabel: stri throw new Error(`${modelLabel} 未返回可用图片:${reason.slice(0, 160)}`); } -function getPublicImageGenerationErrorText(error: unknown) { +function getPublicImageGenerationErrorText(error: unknown, requestId: string) { const message = error instanceof Error ? error.message : String(error || ""); const publicMessage = getUserFacingImageGenerationErrorMessage(error); if (!message || (publicMessage === message && INTERNAL_IMAGE_GENERATION_FAILURE_PATTERN.test(message))) { - return "图片生成失败,请稍后重试"; + return `图片生成失败,请稍后重试。若连续失败,请联系管理员。\n\n请求编号:${requestId}`; } - return publicMessage; + return `${publicMessage}\n\n请求编号:${requestId}`; } @@ -9630,9 +9633,10 @@ export async function POST(request: Request) { // 如果故障转移也失败,返回错误消息 const errorMessageId = generateUUID(); + const errorText = getPublicImageGenerationErrorText(error, errorMessageId); writeAssistantTextChunk( writer, - "图片生成失败,请稍后重试", + errorText, "gemini:error", errorMessageId, ); @@ -9652,7 +9656,7 @@ export async function POST(request: Request) { id: errorMessageId, chatId: id, role: "assistant", - parts: [{ type: "text", text: "图片生成失败,请稍后重试" }], + parts: [{ type: "text", text: errorText }], attachments: [], createdAt: new Date(), videoStorageStatus: null, @@ -10332,7 +10336,7 @@ export async function POST(request: Request) { // 返回错误消息 const errorMessageId = generateUUID(); - const errorText = getPublicImageGenerationErrorText(error); + const errorText = getPublicImageGenerationErrorText(error, errorMessageId); writeAssistantTextChunk( writer, errorText, @@ -10772,9 +10776,10 @@ export async function POST(request: Request) { } catch (error) { chatApiLogger.error("[Chat API] Volcano Vision generation error:", error); const errorMessageId = generateUUID(); + const errorText = getPublicImageGenerationErrorText(error, errorMessageId); writeAssistantTextChunk( writer, - "图片生成失败,请稍后重试", + errorText, "volcano:error", errorMessageId, ); @@ -10794,7 +10799,7 @@ export async function POST(request: Request) { id: errorMessageId, chatId: id, role: "assistant", - parts: [{ type: "text", text: "图片生成失败,请稍后重试" }], + parts: [{ type: "text", text: errorText }], attachments: [], // 必须提供 attachments 字段 createdAt: new Date(), videoStorageStatus: null, @@ -11702,7 +11707,7 @@ export async function POST(request: Request) { chatApiLogger.error("[Chat API] OpenAI-compatible image model generation error:", error); const errorMessageId = generateUUID(); - const errorText = getPublicImageGenerationErrorText(error); + const errorText = getPublicImageGenerationErrorText(error, errorMessageId); dataStream.write({ type: "data-appendMessage", diff --git a/app/api/canvas/[id]/capabilities/route.ts b/app/api/canvas/[id]/capabilities/route.ts index 9507c77d9..f146da491 100644 --- a/app/api/canvas/[id]/capabilities/route.ts +++ b/app/api/canvas/[id]/capabilities/route.ts @@ -8,6 +8,7 @@ import { getCanvasCapabilityManifest } from "@/lib/canvas/capabilities/registry" import { executeCanvasCapabilityRuntime } from "@/lib/canvas/capabilities/runtime"; import { getCanvasCapabilityTask } from "@/lib/canvas/capabilities/task-store"; import { CanvasAgentMutationError } from "@/lib/canvas/agent-mutations"; +import { formatRequestFailure, getRequestFailureDetails } from "@/lib/request-error"; export const dynamic = "force-dynamic"; export const maxDuration = 600; @@ -39,7 +40,8 @@ export async function GET( { status: 404 } ); } - return NextResponse.json({ success: true, task }); + const failure = task.status === "failed" ? getRequestFailureDetails(task.error) : undefined; + return NextResponse.json({ success: true, task: { ...task, failure, error: failure ? formatRequestFailure(failure, task.id) : task.error } }); } export async function POST( @@ -51,9 +53,11 @@ export async function POST( return NextResponse.json({ success: false, error: "未登录" }, { status: 401 }); } + let requestId: string | undefined; try { const { id } = await params; const command = canvasCapabilityClientCommandSchema.parse(await request.json()); + requestId = command.requestId; const result = await executeCanvasCapabilityRuntime({ command, projectId: id, @@ -67,8 +71,9 @@ export async function POST( error instanceof CanvasCapabilityRuntimeError || error instanceof CanvasAgentMutationError ) { + const failure = getRequestFailureDetails(error); return NextResponse.json( - { success: false, error: error.message, code: error.code }, + { success: false, error: formatRequestFailure(failure, requestId), code: error.code, failure, requestId }, { status: error.status } ); } @@ -82,9 +87,9 @@ export async function POST( { status: 400 } ); } - const message = error instanceof Error ? error.message : "执行画布能力失败"; + const failure = getRequestFailureDetails(error, "执行画布能力失败"); return NextResponse.json( - { success: false, error: message }, + { success: false, error: formatRequestFailure(failure, requestId), failure, requestId }, { status: 500 } ); } diff --git a/app/api/canvas/generate/image/route.ts b/app/api/canvas/generate/image/route.ts index 5060327d3..835f4a6a5 100644 --- a/app/api/canvas/generate/image/route.ts +++ b/app/api/canvas/generate/image/route.ts @@ -2,6 +2,7 @@ import { executeCanvasImageResponse } from "@/lib/canvas/generate-image-response import { NextResponse } from "next/server"; import { auth } from "@/app/(auth)/auth"; import { getCanvasImageTaskForUser } from "@/lib/canvas/image-task-queue"; +import { formatRequestFailure, getRequestFailureDetails } from "@/lib/request-error"; export const maxDuration = 1200; @@ -23,13 +24,15 @@ export async function GET(request: Request) { return NextResponse.json({ success: false, error: "任务不存在" }, { status: 404 }); } + const failure = task.status === "failed" ? getRequestFailureDetails(task.error) : undefined; return NextResponse.json({ success: true, task: { id: task.id, status: task.status, attempts: task.attempts, - error: task.error || null, + error: failure ? formatRequestFailure(failure, task.id) : null, + failure, result: task.result || null, createdAt: task.createdAt, startedAt: task.startedAt, diff --git a/lib/canvas/capabilities/runtime.ts b/lib/canvas/capabilities/runtime.ts index 84ab4da92..ff5a61dc9 100644 --- a/lib/canvas/capabilities/runtime.ts +++ b/lib/canvas/capabilities/runtime.ts @@ -14,13 +14,16 @@ import { getCanvasCapability } from "@/lib/canvas/capabilities/registry"; import { enqueueCanvasCapabilityExecution } from "@/lib/canvas/capabilities/task-queue"; import { createOrRecoverCanvasCapabilityTask, - MAX_CANVAS_CAPABILITY_TASK_ATTEMPTS, + getCanvasCapabilityTask, + heartbeatCanvasCapabilityTask, markCanvasCapabilityTaskFailed, markCanvasCapabilityTaskRunning, markCanvasCapabilityTaskSucceeded, } from "@/lib/canvas/capabilities/task-store"; import { applyCanvasOperationPlan } from "@/lib/canvas/agent-mutations"; import type { CanvasOperationPlan } from "@/types/canvas/agent-protocol"; +import { formatRequestFailure, getRequestFailureDetails } from "@/lib/request-error"; +import { trackRuntimeWork } from "@/lib/observability/active-work"; export type CanvasCapabilityRuntimeResult = { clientCommand?: CanvasCapabilityClientCommand; @@ -58,9 +61,11 @@ function serializeTask(task: { status: string; version: number; }) { + const failure = task.status === "failed" ? getRequestFailureDetails(task.error) : undefined; return { attempts: task.attempts, - error: task.error, + error: failure ? formatRequestFailure(failure, task.id) : task.error, + failure, id: task.id, requestId: task.requestId, result: task.result, @@ -69,6 +74,13 @@ function serializeTask(task: { }; } +function storedTaskResult(task: Parameters[0], mode: "headless" | "interactive"): CanvasCapabilityRuntimeResult { + const result = task.status === "succeeded" && task.result && typeof task.result === "object" + ? task.result as Omit + : { mode }; + return { ...result, task: serializeTask(task) }; +} + function buildMediaOperationPlan(params: { command: CanvasCapabilityClientCommand; mediaResults: CanvasCapabilityMediaResult[]; @@ -211,11 +223,7 @@ export async function executeCanvasCapabilityRuntime(params: { }); if (task.status === "succeeded") { - const storedResult = - task.result && typeof task.result === "object" - ? (task.result as Omit) - : ({ mode: "headless" } as const); - return { ...storedResult, task: serializeTask(task) }; + return storedTaskResult(task, "headless"); } if (task.status === "running") { return { @@ -225,15 +233,8 @@ export async function executeCanvasCapabilityRuntime(params: { task: serializeTask(task), }; } - if ( - task.status === "failed" && - task.attempts >= MAX_CANVAS_CAPABILITY_TASK_ATTEMPTS - ) { - throw new CanvasCapabilityRuntimeError( - task.error || "画布能力任务超过最大重试次数", - 409, - "CANVAS_CAPABILITY_RETRY_EXHAUSTED" - ); + if (task.status === "failed") { + return { mode: "headless", task: serializeTask(task) }; } const shouldRunHeadlessly = shouldExecuteCanvasCapabilityHeadlessly({ @@ -241,7 +242,9 @@ export async function executeCanvasCapabilityRuntime(params: { command, }); if (!shouldRunHeadlessly) { - task = await markCanvasCapabilityTaskRunning(task.id); + const claimed = await markCanvasCapabilityTaskRunning(task.id); + if (!claimed) return storedTaskResult((await getCanvasCapabilityTask({ requestId: command.requestId, userId: params.userId, projectId: params.projectId })) || task, "interactive"); + task = claimed; const result = { mode: "interactive" as const, clientCommand: command, @@ -250,10 +253,18 @@ export async function executeCanvasCapabilityRuntime(params: { return { ...result, task: serializeTask(task) }; } - return enqueueCanvasCapabilityExecution({ + const heartbeat = setInterval(() => { + void heartbeatCanvasCapabilityTask(task.id).catch((error) => { + console.warn("[Canvas Capability] Task heartbeat failed:", error); + }); + }, 30_000); + try { + return await trackRuntimeWork(() => enqueueCanvasCapabilityExecution({ projectId: params.projectId, execute: async () => { - task = await markCanvasCapabilityTaskRunning(task.id); + const claimed = await markCanvasCapabilityTaskRunning(task.id); + if (!claimed) return storedTaskResult((await getCanvasCapabilityTask({ requestId: command.requestId, userId: params.userId, projectId: params.projectId })) || task, "headless"); + task = claimed; try { const projectRevision = context.project.updatedAt.toISOString(); const mutationPlan = buildMutationOperationPlan({ @@ -299,5 +310,8 @@ export async function executeCanvasCapabilityRuntime(params: { throw error; } }, - }); + })); + } finally { + clearInterval(heartbeat); + } } diff --git a/lib/canvas/capabilities/task-store.ts b/lib/canvas/capabilities/task-store.ts index db96249bb..9fa17eb3c 100644 --- a/lib/canvas/capabilities/task-store.ts +++ b/lib/canvas/capabilities/task-store.ts @@ -1,11 +1,11 @@ -import { and, eq, lt, sql } from "drizzle-orm"; +import { and, eq, inArray, lt, sql } from "drizzle-orm"; import type { CanvasCapabilityClientCommand } from "@/lib/canvas/capabilities/client-command"; import { db } from "@/lib/db"; import { canvasCapabilityTask } from "@/lib/db/schema"; const STALE_RUNNING_TASK_MS = 15 * 60 * 1000; -export const MAX_CANVAS_CAPABILITY_TASK_ATTEMPTS = 3; +const INTERRUPTED_TASK_MESSAGE = "任务执行中断,结果尚未确认。请先查看画布和历史记录,或联系管理员核对原任务,避免重复处理。"; export type CanvasCapabilityTaskResult = Record; @@ -25,7 +25,25 @@ export async function getCanvasCapabilityTask(params: { ) ) .limit(1); - return task || null; + if (!task) return null; + const staleBefore = new Date(Date.now() - STALE_RUNNING_TASK_MS); + if (["pending", "running"].includes(task.status) && task.updatedAt < staleBefore) { + const [interrupted] = await db.update(canvasCapabilityTask).set({ + status: "failed", + error: INTERRUPTED_TASK_MESSAGE, + completedAt: new Date(), + updatedAt: new Date(), + }).where(and( + eq(canvasCapabilityTask.id, task.id), + inArray(canvasCapabilityTask.status, ["pending", "running"]), + lt(canvasCapabilityTask.updatedAt, staleBefore), + )).returning(); + if (interrupted) return interrupted; + const [latest] = await db.select().from(canvasCapabilityTask) + .where(eq(canvasCapabilityTask.id, task.id)).limit(1); + return latest || task; + } + return task; } export async function createOrRecoverCanvasCapabilityTask(params: { @@ -42,6 +60,7 @@ export async function createOrRecoverCanvasCapabilityTask(params: { capabilityId: params.command.capabilityId, source: params.command.source, request: params.command, + updatedAt: new Date(), }) .onConflictDoNothing({ target: canvasCapabilityTask.requestId }); @@ -50,59 +69,19 @@ export async function createOrRecoverCanvasCapabilityTask(params: { userId: params.userId, projectId: params.projectId, }; - let task = await getCanvasCapabilityTask(taskLookup); + const task = await getCanvasCapabilityTask(taskLookup); if (!task) { throw new Error("无法创建画布能力任务"); } - const staleBefore = new Date(Date.now() - STALE_RUNNING_TASK_MS); - if ( - task.status === "failed" && - task.attempts < MAX_CANVAS_CAPABILITY_TASK_ATTEMPTS - ) { - await db - .update(canvasCapabilityTask) - .set({ - status: "pending", - error: null, - completedAt: null, - updatedAt: new Date(), - }) - .where(eq(canvasCapabilityTask.id, task.id)); - } else if ( - task.status === "running" && - task.updatedAt < staleBefore && - task.attempts < MAX_CANVAS_CAPABILITY_TASK_ATTEMPTS - ) { - await db - .update(canvasCapabilityTask) - .set({ - status: "pending", - error: "恢复超时的执行中任务", - updatedAt: new Date(), - }) - .where( - and( - eq(canvasCapabilityTask.id, task.id), - lt(canvasCapabilityTask.updatedAt, staleBefore) - ) - ); - } else if (task.status === "running" && task.updatedAt < staleBefore) { - await db - .update(canvasCapabilityTask) - .set({ - status: "failed", - error: "画布能力任务超过最大重试次数", - completedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(canvasCapabilityTask.id, task.id)); - } - - task = (await getCanvasCapabilityTask(taskLookup)) || task; return task; } +export async function heartbeatCanvasCapabilityTask(taskId: string) { + await db.update(canvasCapabilityTask).set({ updatedAt: new Date() }) + .where(and(eq(canvasCapabilityTask.id, taskId), inArray(canvasCapabilityTask.status, ["pending", "running"]))); +} + export async function markCanvasCapabilityTaskRunning(taskId: string) { const now = new Date(); const [task] = await db @@ -115,11 +94,10 @@ export async function markCanvasCapabilityTaskRunning(taskId: string) { error: null, updatedAt: now, }) - .where(eq(canvasCapabilityTask.id, taskId)) + .where(and(eq(canvasCapabilityTask.id, taskId), eq(canvasCapabilityTask.status, "pending"))) .returning(); - if (!task) throw new Error("画布能力任务不存在"); - return task; + return task || null; } export async function markCanvasCapabilityTaskSucceeded( diff --git a/lib/canvas/generate-image-response.ts b/lib/canvas/generate-image-response.ts index 0762ef2be..09f85b7db 100644 --- a/lib/canvas/generate-image-response.ts +++ b/lib/canvas/generate-image-response.ts @@ -10,7 +10,7 @@ import { generateCanvasImage, type CanvasGenerateImageRequest } from "@/lib/canv import { getPointsSystemConfig } from "@/lib/db/queries"; import { prepareFeaturePointsCharge } from "@/lib/points/feature-charge"; import { getImageToolCreditCost, type ImageToolCreditFeature } from "@/lib/points/config"; -import { getUserFacingImageGenerationErrorMessage } from "@/lib/request-error"; +import { getUserFacingImageGenerationErrorMessage, getRequestFailureDetails } from "@/lib/request-error"; interface GenerateImageRequest extends CanvasGenerateImageRequest { async?: boolean; @@ -121,7 +121,7 @@ export async function executeCanvasImageResponse(body: GenerateImageRequest, use } catch (error) { console.error("[Canvas Generate Image] Error:", error); return NextResponse.json( - { success: false, error: getUserFacingImageGenerationErrorMessage(error), + { success: false, error: getUserFacingImageGenerationErrorMessage(error), failure: getRequestFailureDetails(error), ...(error instanceof ProviderTaskUncertainError ? { code: "PROVIDER_TASK_UNCERTAIN" } : {}) }, { status: error instanceof ProviderTaskUncertainError ? 503 : 500 }, ); diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index 8c926fd4f..e5cdde04a 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -38,6 +38,7 @@ import { } from "@/lib/agents/detail-replacement-batch"; import { executeDetailReplacementBatch } from "@/lib/agents/detail-replacement-batch-runner"; import { getUserFacingImageGenerationErrorMessage } from "@/lib/request-error"; +import { trackRuntimeWork } from "@/lib/observability/active-work"; type CanvasImageTaskStatus = | "waiting" @@ -328,7 +329,7 @@ async function markStaleDatabaseRunningTasksFailed() { .update(canvasImageTask) .set({ status: "failed", - error: "图片生成任务已超时,请重新发起生成", + error: "图片任务执行中断或等待超时,结果尚未确认。请先查看原任务和历史记录,联系管理员核对,避免重复生成。", completedAt, updatedAt: completedAt, }) @@ -383,7 +384,7 @@ async function runSingleTask(entry: PendingTaskEntry) { } } const runningTimeoutMs = getCanvasImageRunningTimeoutMs(request); - const execution: Promise = isStyleProTaskRequest(request) + const execution: Promise = trackRuntimeWork(() => isStyleProTaskRequest(request) ? executeStyleProTask({ request, userId: String(task.userId || entry.userId), @@ -412,7 +413,7 @@ async function runSingleTask(entry: PendingTaskEntry) { ownerKey, request: imageRequest, }); - })(); + })()); const heartbeat = isDetailReplacementBatchRequest(request) ? setInterval(() => { void db.update(canvasImageTask).set({ updatedAt: new Date() }) @@ -426,7 +427,7 @@ async function runSingleTask(entry: PendingTaskEntry) { try { result = await withCanvasImageTaskExecutionTimeout( execution, - runningTimeoutMs, + Math.max(runningTimeoutMs, LOCAL_ACTIVE_RUNNING_HARD_TIMEOUT_MS), entry.taskId ); } finally { @@ -751,7 +752,7 @@ export async function getCanvasImageTaskForUser(params: { if (shouldMarkStale) { const staleBefore = new Date(now - runningTimeoutMs); - const timeoutMessage = "图片生成任务已超时,请重新发起生成"; + const timeoutMessage = "图片任务执行中断或等待超时,结果尚未确认。请先查看原任务和历史记录,联系管理员核对,避免重复生成。"; const completedAt = new Date(); const [timedOutTask] = await db .update(canvasImageTask) diff --git a/lib/client/canvas-toolbar-capability.ts b/lib/client/canvas-toolbar-capability.ts index b3980057d..431d71350 100644 --- a/lib/client/canvas-toolbar-capability.ts +++ b/lib/client/canvas-toolbar-capability.ts @@ -1,6 +1,40 @@ import type { CanvasCapabilityClientCommand } from "@/lib/canvas/capabilities/client-command"; import { mergeCanvasSnapshots } from "@/lib/video-canvas/merge-snapshot"; import { useCanvasStore } from "@/store/canvas"; +import { formatRequestFailure, getRequestFailureDetails } from "@/lib/request-error"; + +async function requestCapabilityResult(command: CanvasCapabilityClientCommand) { + const endpoint = `/api/canvas/${command.projectId}/capabilities`; + let payload: any = null; + try { + const response = await fetch(endpoint, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(command), + }); + payload = await response.json().catch(() => null); + if (payload?.success === false || (!response.ok && ![502, 503, 504, 524].includes(response.status))) { + throw Object.assign(new Error(payload?.error || "图片处理请求未通过"), { confirmedRejection: true }); + } + if (!response.ok) payload = null; + } catch (error) { + if ((error as { confirmedRejection?: boolean })?.confirmedRejection) throw error; + } + const deadline = Date.now() + 20 * 60 * 1000; + while (Date.now() < deadline) { + const task = payload?.task; + if (task?.status === "succeeded") return payload; + if (task?.status === "failed") throw new Error(task.error || "图片处理失败,请联系管理员核对任务结果"); + await new Promise(resolve => setTimeout(resolve, 1500)); + try { + const response = await fetch(`${endpoint}?requestId=${encodeURIComponent(command.requestId)}`, { cache: "no-store", signal: AbortSignal.timeout(12_000) }); + payload = response.ok ? await response.json() : null; + } catch { + payload = null; + } + } + throw new Error(formatRequestFailure(getRequestFailureDetails({ providerOutcomeUnknown: true }), command.requestId)); +} /** Runs after the editor has saved and while its save queue is reserved. */ export async function executeCanvasToolbarCapability( @@ -10,15 +44,7 @@ export async function executeCanvasToolbarCapability( const before = useCanvasStore.getState(); if (before.projectId !== projectId) throw new Error("画布已切换,请重新选择图片"); const base = JSON.stringify({ canvasData: before.getCanvasState() }); - const response = await fetch(`/api/canvas/${projectId}/capabilities`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify(command), - }); - const payload = await response.json().catch(() => ({})); - if (!response.ok || payload?.success !== true) { - throw new Error(payload?.error || "图片处理失败"); - } + await requestCapabilityResult(command); const projectResponse = await fetch(`/api/canvas/${projectId}`); const result = await projectResponse.json(); diff --git a/lib/jobs/image-archive-job.ts b/lib/jobs/image-archive-job.ts index b62d2a94a..10ce21f32 100644 --- a/lib/jobs/image-archive-job.ts +++ b/lib/jobs/image-archive-job.ts @@ -1,5 +1,6 @@ import { preparePendingImageDownloads } from "@/lib/storage/image-download-preparation"; import "server-only"; +import { trackRuntimeWork } from "@/lib/observability/active-work"; import { archiveTemporaryImage } from "@/lib/storage/image-archive"; import { @@ -26,6 +27,7 @@ function getBatchSize() { } export async function runImageArchiveJob() { + return trackRuntimeWork(async () => { const batchSize = getBatchSize(); console.log(`[ImageArchiveJob] Start image archive job (batchSize=${batchSize})`); @@ -125,4 +127,5 @@ export async function runImageArchiveJob() { downloadTasks, pendingDownloads: await hasPendingImageDownloads(), }; + }); } diff --git a/lib/observability/active-work.ts b/lib/observability/active-work.ts new file mode 100644 index 000000000..8e5f3c093 --- /dev/null +++ b/lib/observability/active-work.ts @@ -0,0 +1,23 @@ +import "server-only"; +import { unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const state = globalThis as typeof globalThis & { __aiChatbotActiveWork?: number }; +const path = join(process.platform === "win32" ? tmpdir() : "/tmp", `ai-chatbot-active-work-${process.pid}.json`); + +function update(delta: number) { + state.__aiChatbotActiveWork = (state.__aiChatbotActiveWork || 0) + delta; + try { + writeFileSync(path, JSON.stringify({ pid: process.pid, active: state.__aiChatbotActiveWork }), { mode: 0o600 }); + } catch (error) { + try { unlinkSync(path); } catch {} + console.warn("[Active Work] Failed to record running work:", error); + } +} + +/** Includes generation that continues after the client connection closes. */ +export async function trackRuntimeWork(execute: () => T | Promise): Promise { + update(1); + try { return await execute(); } finally { update(-1); } +} diff --git a/lib/request-error.ts b/lib/request-error.ts index 3a060050f..c18c23167 100644 --- a/lib/request-error.ts +++ b/lib/request-error.ts @@ -189,14 +189,6 @@ export function isGatewayResponseTimeout(error: unknown): boolean { /\bHTTP\s*524\b|\b524\s*:\s*A timeout occurred\b|Error code\s*524\b/i.test(getRequestErrorText(error)); } -function getPublicGatewayErrorMessage(error: unknown): string | null { - if (isGatewayResponseTimeout(error)) { - return "模型服务响应超时(HTTP 524),生成结果尚未确认。请先查看历史记录,避免重复提交。"; - } - const message = getRequestErrorMessage(error); - return message ? getGatewayErrorPageMessage(message) : null; -} - export function isProviderBalanceError(error: unknown): boolean { const message = getRequestErrorText(error); return PROVIDER_BALANCE_PATTERNS.some((pattern) => pattern.test(message)); @@ -221,19 +213,74 @@ export function isRateLimitError(error: unknown): boolean { RATE_LIMIT_PATTERNS.some((pattern) => pattern.test(message)); } +export type RequestFailureDetails = { + code: string; + message: string; + recommendation: string; + outcome: "failed" | "unknown"; +}; + +export function getRequestFailureDetails( + error: unknown, + fallbackMessage = "请求未能完成", +): RequestFailureDetails { + const text = getRequestErrorText(error); + const status = getRequestErrorStatus(error); + const detail = (code: string, message: string, recommendation: string, outcome: "failed" | "unknown" = "failed") => + ({ code, message, recommendation, outcome }); + if (isGatewayResponseTimeout(error)) { + return detail("RESULT_UNCONFIRMED", "模型服务响应超时(HTTP 524),生成结果尚未确认。", "请先查看历史记录,避免重复提交。", "unknown"); + } + if ((isRecord(error) && error.providerOutcomeUnknown === true) || /结果(?:尚)?未确认|结果暂未确认|request_outcome_unknown|任务执行中断|may still be running/i.test(text)) { + return detail("RESULT_UNCONFIRMED", "任务结果尚未确认,可能仍在处理。", "请先查看原任务或历史记录;持续没有结果时联系管理员,避免重复生成。", "unknown"); + } + if (isProviderBalanceError(error)) { + return detail("PROVIDER_BALANCE_EXHAUSTED", "图片服务商线路余额或额度不足,与您的积分余额无关。", "请切换可用模型或联系管理员处理线路额度。"); + } + if (isRateLimitError(error)) { + return detail("RATE_LIMITED", "当前请求过于频繁。", "请稍后重试,避免连续点击;模型线路限流时可切换其他可用模型。"); + } + if (/content.?polic|moderation|safety|policy_rejected|内容(?:敏感|审核|限制)|不能生成像那样|不能生成那样|违规/i.test(text)) { + return detail("CONTENT_POLICY_REJECTED", "模型服务因内容审核限制拒绝了本次请求。", "请调整描述或参考图中的受限内容后再提交;重复提交相同内容通常无法解决。"); + } + if (/上传到存储桶失败|存储.*失败|归档.*失败|archive.*failed|storage.*failed/i.test(text)) { + return detail("RESULT_STORAGE_FAILED", "处理结果保存失败,生成结果可能已经存在。", "请保留当前页面并联系管理员提供请求编号,先不要重复生成。", "unknown"); + } + if (/参考图.*(?:下载失败|不可|失效)|无法获取图片|图片(?:源文件|链接).*不可|获取图片.*失败|get image url failed/i.test(text)) { + return detail("SOURCE_IMAGE_UNAVAILABLE", "请求中的图片无法读取,链接可能已失效或没有访问权限。", "请重新上传原图后再提交;若原图仍可打开,请联系管理员检查取图服务。"); + } + if (/版本冲突|画布.*(?:更新|过期)|CANVAS_REVISION_CONFLICT|stale.canvas/i.test(text)) { + return detail("CANVAS_REVISION_CONFLICT", "画布已更新,本次保存与当前版本冲突。", "请先重新同步画布,再继续操作,避免覆盖较新的修改。"); + } + if (/no_eligible_route|service_unavailable|service_rejected|channel.*unavailable|暂无可用服务|未配置.*模型|没有可用.*(?:模型|通道)|通道.*(?:不可用|未通过)|鉴权/i.test(text)) { + return detail("PROVIDER_UNAVAILABLE", "当前模型服务或通道暂不可用。", "请切换其他可用模型;若多个模型都失败,请联系管理员检查服务配置。"); + } + if (/neither a task ID nor output URLs|已完成.*未取回|completed without.*output/i.test(text)) { + return detail("RESULT_UNCONFIRMED", "模型服务没有返回可读取的任务结果,生成状态尚未确认。", "请保留当前任务并联系管理员核对,避免重复生成。", "unknown"); + } + if (/service_error|upstream_server_error|excessive system load|generate(?: image)? failed/i.test(text)) { + return detail("PROVIDER_EXECUTION_FAILED", "模型服务未能完成本次处理。", "请稍后重试;若连续失败,请切换模型或联系管理员,并提供请求编号。"); + } + if (/timeout|timed out|超时|fetch failed|ECONNRESET|ECONNREFUSED|failed to fetch|networkerror|load failed|network request failed/i.test(text) || [502, 504].includes(status || 0)) { + return detail("RESULT_UNCONFIRMED", "请求连接中断或等待超时,任务结果尚未确认。", "请先查看原任务或历史记录,避免重复提交;持续没有结果时联系管理员。", "unknown"); + } + if (/generate(?: image)? failed|generation failed|upstream_server_error|service_error|excessive system load|未返回可用图片|任务FAILED/i.test(text) || (status !== null && status >= 500)) { + return detail("PROVIDER_EXECUTION_FAILED", "模型服务未能完成本次处理。", "请稍后重试;若连续失败,请切换模型或联系管理员,并提供请求编号。"); + } + const message = getRequestErrorMessage(error); + return detail("REQUEST_FAILED", message && !/^\s*(); function isHighFrequencyReadApi(pathname: string, method: string) { + if (method === "POST" && /^\/api\/canvas\/[^/]+\/image-archives$/.test(pathname)) return true; if (method === "GET" && pathname === "/api/canvas/download-zip-task") return true; if (method !== "GET") return false; if (/^\/api\/canvas\/[^/]+\/detail-archives$/.test(pathname)) return true; @@ -668,7 +669,20 @@ export async function proxy(request: NextRequest, event: NextFetchEvent) { } ); } - } else if (isHighFrequencyReadApi(pathname, request.method)) { + } else if (request.method === "POST" && /^\/api\/canvas\/[^/]+\/image-archives$/.test(pathname)) { + const token = await resolveToken(); + const identifier = typeof token?.sub === "string" && token.sub + ? `user:${token.sub}:image-archives` + : `ip:${clientIP}:image-archives`; + const result = checkRateLimitByIdentifier(identifier, rateLimiters.uiPolling); + if (!result.allowed) { + const retryAfter = Math.max(1, Math.ceil((result.resetTime - Date.now()) / 1000)); + return withAccessLog(NextResponse.json( + { error: "图片状态查询过于频繁,请稍后再查看", retryAfter }, + { status: 429, headers: { "Retry-After": String(retryAfter) } }, + ), { outcome: "rate_limited", severity: "watch", reasons: ["图片归档状态查询限流"] }); + } + } else if (isHighFrequencyReadApi(pathname, request.method)) { const pollingRateLimit = checkRateLimit(request, rateLimiters.uiPolling); if (!pollingRateLimit.allowed) { console.warn(`[Security] UI polling rate limit exceeded for IP: ${clientIP}`, { diff --git a/tests/routes/canvas-capability-task-recovery.test.ts b/tests/routes/canvas-capability-task-recovery.test.ts new file mode 100644 index 000000000..6eaab31c9 --- /dev/null +++ b/tests/routes/canvas-capability-task-recovery.test.ts @@ -0,0 +1,65 @@ +import { expect, test } from "@playwright/test"; +import { serverLoader } from "../helpers/canvas-download-server-loader"; + +function harness(status = "running") { + let now = 2_000_000; + const row: any = { id: "task-a", requestId: "request-a", userId: "user-a", projectId: "project-a", status, attempts: status === "pending" ? 0 : 1, updatedAt: new Date(now - 16 * 60_000), error: null }; + const matches = (predicate: any): boolean => !predicate || (predicate[0] === "and" ? predicate.slice(1).every(matches) + : predicate[0] === "eq" ? row[predicate[1]] === predicate[2] + : predicate[0] === "lt" ? row[predicate[1]] < predicate[2] + : predicate[2].includes(row[predicate[1]])); + const db = { + select: () => ({ from: () => ({ where: (p: any) => ({ limit: async () => matches(p) ? [{ ...row }] : [] }) }) }), + insert: () => ({ values: () => ({ onConflictDoNothing: async () => undefined }) }), + update: () => ({ set: (values: any) => ({ where: (p: any) => { + const apply = () => { + if (!matches(p)) return []; + for (const [key, value] of Object.entries(values)) row[key] = key === "attempts" ? row.attempts + 1 : value; + return [{ ...row }]; + }; + const result = apply(); + return Object.assign(Promise.resolve(result), { returning: async () => result }); + } }) }), + }; + const columns = Object.fromEntries(["id", "requestId", "userId", "projectId", "status", "updatedAt", "attempts"].map(k => [k, k])); + const store = serverLoader({ + "@/lib/db": { db }, "@/lib/db/schema": { canvasCapabilityTask: columns }, + "drizzle-orm": { eq: (...v: any[]) => ["eq", ...v], lt: (...v: any[]) => ["lt", ...v], inArray: (...v: any[]) => ["in", ...v], and: (...v: any[]) => ["and", ...v], sql: () => "increment" }, + Date: class extends Date { constructor(value?: any) { super(value ?? now); } static now() { return now; } }, + })("lib/canvas/capabilities/task-store.ts"); + const lookup = { requestId: "request-a", userId: "user-a", projectId: "project-a" }; + return { row, store, lookup, advance: (ms: number) => { now += ms; } }; +} + +test("a stale task stops reporting running without authorizing another execution", async () => { + const h = harness(); + const task = await h.store.getCanvasCapabilityTask(h.lookup); + expect(task.status).toBe("failed"); + expect(task.error).toContain("结果尚未确认"); + const repeated = await h.store.createOrRecoverCanvasCapabilityTask({ ...h.lookup, command: { requestId: "request-a" } }); + expect(repeated.status).toBe("failed"); + expect(repeated.attempts).toBe(1); +}); + +test("heartbeats keep a legitimate long-running task active beyond fifteen minutes", async () => { + const h = harness(); + await h.store.heartbeatCanvasCapabilityTask("task-a"); + h.advance(14 * 60_000); + await h.store.heartbeatCanvasCapabilityTask("task-a"); + h.advance(14 * 60_000); + expect((await h.store.getCanvasCapabilityTask(h.lookup)).status).toBe("running"); + expect(h.row.attempts).toBe(1); +}); + +test("only the pending task can be claimed and a duplicate claim cannot run again", async () => { + const h = harness("pending"); + expect(await h.store.markCanvasCapabilityTaskRunning("task-a")).toMatchObject({ status: "running", attempts: 1 }); + expect(await h.store.markCanvasCapabilityTaskRunning("task-a")).toBeNull(); + expect(h.row.attempts).toBe(1); +}); + +test("another user cannot inspect or change a stale task", async () => { + const h = harness(); + expect(await h.store.getCanvasCapabilityTask({ ...h.lookup, userId: "user-b" })).toBeNull(); + expect(h.row.status).toBe("running"); +}); diff --git a/tests/routes/canvas-toolbar-capability.test.ts b/tests/routes/canvas-toolbar-capability.test.ts index c5d7d72ee..907502d19 100644 --- a/tests/routes/canvas-toolbar-capability.test.ts +++ b/tests/routes/canvas-toolbar-capability.test.ts @@ -47,7 +47,7 @@ test("merges the result with edits made while processing and advances the save r globalThis.fetch = (async (url) => { if (String(url).endsWith("/capabilities")) { useCanvasStore.getState().moveNode("source", { x: 50, y: 80 }); - return Response.json({ success: true }); + return Response.json({ success: true, task: { status: "succeeded" } }); } return Response.json({ success: true, data: { updatedAt: "r2", canvasData: { nodes: [node("source"), node("result")], edges: [] }, @@ -67,12 +67,31 @@ test("a provider failure is shown without reloading or changing the canvas", asy expect(useCanvasStore.getState().canvasRevision).toBe("r1"); }); +test("a lost submission response polls the original task and never resubmits", async () => { + const requests: { method: string; url: string }[] = []; + const originalSetTimeout = globalThis.setTimeout; + let polls = 0; + globalThis.setTimeout = ((fn: () => void) => originalSetTimeout(fn, 0)) as typeof setTimeout; + globalThis.fetch = (async (url, init) => { + requests.push({ method: init?.method || "GET", url: String(url) }); + if (init?.method === "POST") throw new TypeError("fetch failed"); + if (String(url).includes("/capabilities?")) return Response.json({ success: true, task: { status: ++polls === 1 ? "running" : "succeeded" } }); + return Response.json({ success: true, data: { updatedAt: "r2", canvasData: { nodes: [node("source"), node("result")], edges: [] } } }); + }) as typeof fetch; + const submitted = command(); + try { await executeCanvasToolbarCapability(submitted); } finally { globalThis.setTimeout = originalSetTimeout; } + expect(requests.filter(r => r.method === "POST")).toHaveLength(1); + expect(requests.filter(r => r.url.includes("/capabilities?"))).toHaveLength(2); + expect(requests.filter(r => r.url.includes("/capabilities?")).every(r => r.url.includes(encodeURIComponent(submitted.requestId)))).toBe(true); + expect(useCanvasStore.getState().nodes.map(n => n.id)).toEqual(["source", "result"]); +}); + test("switching projects during processing does not load the old result into the new project", async () => { globalThis.fetch = (async (url) => { if (String(url).endsWith("/capabilities")) { useCanvasStore.getState().setProject("project-b"); useCanvasStore.getState().loadCanvas({ nodes: [node("other") as any], edges: [] }); - return Response.json({ success: true }); + return Response.json({ success: true, task: { status: "succeeded" } }); } return Response.json({ success: true, data: { updatedAt: "r2", canvasData: { nodes: [node("result")], edges: [] } } }); }) as typeof fetch; @@ -104,7 +123,7 @@ test("a result merge conflict retains both the local edit and its original revis globalThis.fetch = (async (url) => { if (String(url).endsWith("/capabilities")) { useCanvasStore.getState().moveNode("source", { x: 50, y: 80 }); - return Response.json({ success: true }); + return Response.json({ success: true, task: { status: "succeeded" } }); } return Response.json({ success: true, data: { updatedAt: "r2", canvasData: { nodes: [{ ...node("source"), position: { x: 200, y: 80 } }, node("result")], edges: [], diff --git a/tests/routes/image-provider-error-classification.test.ts b/tests/routes/image-provider-error-classification.test.ts index 5e3d93e71..8dff2cae6 100644 --- a/tests/routes/image-provider-error-classification.test.ts +++ b/tests/routes/image-provider-error-classification.test.ts @@ -2,11 +2,32 @@ import { expect, test } from "@playwright/test"; import { getUserFacingImageGenerationErrorMessage, getUserFacingRequestErrorMessage, + getRequestFailureDetails, + formatRequestFailure, getGatewayErrorPageMessage, isGatewayResponseTimeout, isProviderBalanceError, isRateLimitError, } from "@/lib/request-error"; + +test("known provider and platform failures explain their cause and next action", () => { + for (const [message, code, next] of [ + ["We are sorry, this prompt may violate our content policies", "CONTENT_POLICY_REJECTED", "调整描述"], + ["The channel is temporarily unavailable", "PROVIDER_UNAVAILABLE", "切换其他"], + ["generate image failed", "PROVIDER_EXECUTION_FAILED", "稍后重试"], + ["excessive system load", "PROVIDER_EXECUTION_FAILED", "稍后重试"], + ["get image url failed", "SOURCE_IMAGE_UNAVAILABLE", "重新上传"], + ["图片上传到存储桶失败", "RESULT_STORAGE_FAILED", "不要重复生成"], + ["任务执行中断,结果尚未确认", "RESULT_UNCONFIRMED", "避免重复生成"], + ]) { + const failure = getRequestFailureDetails(new Error(message)); + expect(failure.code).toBe(code); + expect(failure.recommendation).toContain(next); + expect(formatRequestFailure(failure, "request-123")).toContain("请求编号:request-123"); + } + expect(getRequestFailureDetails({ providerOutcomeUnknown: true, message: "content policy" }).outcome).toBe("unknown"); + expect(getRequestFailureDetails(new Error("图片上传到存储桶失败")).outcome).toBe("unknown"); +}); import { isImageModelBalanceBlocked, resolveConfiguredImageBalanceFallback } from "@/lib/ai/runtime/image-balance"; import { serverLoader } from "../helpers/canvas-download-server-loader"; diff --git a/tests/routes/runtime-active-work.test.ts b/tests/routes/runtime-active-work.test.ts new file mode 100644 index 000000000..795f49384 --- /dev/null +++ b/tests/routes/runtime-active-work.test.ts @@ -0,0 +1,19 @@ +import { expect, test } from "@playwright/test"; +import { serverLoader } from "../helpers/canvas-download-server-loader"; + +test("release draining includes background work until its original promise settles", async () => { + const counts: number[] = []; + const work = serverLoader({ + "node:fs": { writeFileSync: (_path: string, text: string) => counts.push(JSON.parse(text).active), unlinkSync() {} }, + })("lib/observability/active-work.ts"); + let finish!: () => void; + const original = new Promise(resolve => { finish = resolve; }); + const pending = work.trackRuntimeWork(() => original); + await work.trackRuntimeWork(async () => "another request"); + expect(counts).toEqual([1, 2, 1]); + finish(); + await pending; + expect(counts).toEqual([1, 2, 1, 0]); + await expect(work.trackRuntimeWork(async () => { throw new Error("expected failure"); })).rejects.toThrow("expected failure"); + expect(counts.at(-1)).toBe(0); +});