From 03b8a0deb0e8310128a3854ba26d82d896f0ddd1 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 30 Oct 2026 15:37:23 +0800 Subject: [PATCH] chore(repo): preserve production-48 source snapshot Recover the existing 19 modified tracked files and 6 untracked source files from the production working tree without changing their contents. Production source HEAD: 439fd56fe120fdb9f855b50fa4df9ce2e71bb72a Production build ID: QhoygcZ4-x42PwCgCEj1d Captured: 2026-10-30; all 25 source file SHA256 values verified. This is a source recovery checkpoint; no deployment was performed. --- app/api/canvas/agent-chat/route.ts | 25 +- components/canvas/CanvasChatPanel.tsx | 15 + .../canvas/LiblibVideoCanvasReplica.tsx | 36 +- lib/ai/canvas-agent-runtime/protocol.ts | 3 + lib/ai/canvas-agent-runtime/store.ts | 2 + lib/ai/providers/vip98k.ts | 4 + lib/ai/vip98k-runtime-config.ts | 8 +- lib/ai/vip98k-task-invocation.ts | 39 + lib/ai/zcode-agent/assistant-output.ts | 14 + lib/ai/zcode-agent/model-gateway.ts | 13 +- lib/ai/zcode-agent/task.ts | 26 + lib/canvas/image-task-queue.ts | 2 +- lib/video-canvas/config.ts | 8 +- next-env.d.ts | 2 +- .../storage/audit-aliyun-qiniu-references.cjs | 6 + .../storage/inventory-aliyun-qiniu-source.cjs | 8 + scripts/storage/migrate-aliyun-to-minio.ts | 1308 +++++++++++++++++ scripts/storage/migrate-aliyun-to-qiniu.cjs | 220 +++ .../storage/register-aliyun-qiniu-aliases.cjs | 6 + scripts/zcode-agent/native-fixture.mjs | 2 + scripts/zcode-agent/task-fixture.ts | 26 + services/zcode-agent/runner.d.mts | 2 +- services/zcode-agent/runner.mjs | 5 +- tests/integration/agent-model-gateway.test.ts | 15 + tsconfig.json | 22 +- 25 files changed, 1766 insertions(+), 51 deletions(-) create mode 100644 lib/ai/zcode-agent/assistant-output.ts create mode 100644 scripts/storage/audit-aliyun-qiniu-references.cjs create mode 100644 scripts/storage/inventory-aliyun-qiniu-source.cjs create mode 100644 scripts/storage/migrate-aliyun-to-minio.ts create mode 100644 scripts/storage/migrate-aliyun-to-qiniu.cjs create mode 100644 scripts/storage/register-aliyun-qiniu-aliases.cjs diff --git a/app/api/canvas/agent-chat/route.ts b/app/api/canvas/agent-chat/route.ts index 78758451e..69c930789 100644 --- a/app/api/canvas/agent-chat/route.ts +++ b/app/api/canvas/agent-chat/route.ts @@ -31,6 +31,7 @@ import { saveMessages, } from "@/lib/db/queries"; import { runZcodeCanvasTask } from "@/lib/ai/zcode-agent/task"; +import { canvasAgentReplyText } from "@/lib/ai/zcode-agent/assistant-output"; import { createZcodeModelGateway } from "@/lib/ai/zcode-agent/model-gateway"; import { newAgentRunSnapshot, publicAgentError } from "@/lib/ai/canvas-agent-runtime/protocol"; import { shouldUseCanvasAgentRuntime } from "@/lib/ai/canvas-agent-runtime/routing"; @@ -288,11 +289,11 @@ export async function POST(request: Request) { const textId = crypto.randomUUID(); writer.write({ type: "start", messageId: assistantMessageId }); writer.write({ type: "text-start", id: textId }); - writer.write({ - type: "text-delta", - id: textId, - delta: isVideo ? "正在按步骤处理你的视频任务。\n\n" : "正在按步骤处理你的图片任务。\n\n", - }); + let streamedText = ""; + const emitAssistantText = (delta: string) => { + streamedText += delta; + writer.write({ type: "text-delta", id: textId, delta }); + }; try { const videoPorts = isVideo ? await createVideoRuntimePorts({ @@ -343,6 +344,7 @@ export async function POST(request: Request) { runId: run.id, snapshot, definitions: tools.definitions, instructions: (isVideo ? videoRuntimeSystemPrompt : canvasRuntimeSystemPrompt)(snapshot.objective, snapshot.summary, snapshot.inputs.map(input => input.text), context), context, gateway, signal: activeModelController.signal, + onAssistantText: emitAssistantText, tools: tools.executable, checkpoint: async (state, stateStatus) => { if (leaseLost) throw new AgentRunError("任务执行已中断"); @@ -406,7 +408,7 @@ export async function POST(request: Request) { signal: activeModelController.signal, }), }); - const finalText = snapshot.message; + const finalText = canvasAgentReplyText(snapshot); const progress = canvasAgentProgress(agentRunView({ ...run, snapshot, status })); writer.write({ type: "data-canvasAgentProgress", id: run.id, data: progress }); if (!isVideo) @@ -439,19 +441,20 @@ export async function POST(request: Request) { }, ], }); - writer.write({ type: "text-delta", id: textId, delta: finalText }); + emitAssistantText(finalText.startsWith(streamedText) ? finalText.slice(streamedText.length) : `\n\n${snapshot.message}`); } catch (error) { - const text = publicAgentError(error); - await checkpointAgentRun(run, { ...snapshot, message: text }, "needs_review").catch( + const statusText = publicAgentError(error); + const text = canvasAgentReplyText({ ...snapshot, message: statusText }); + await checkpointAgentRun(run, { ...snapshot, message: statusText }, "needs_review").catch( () => undefined, ); - const progress = canvasAgentProgress(agentRunView({ ...run, snapshot: { ...snapshot, message: text }, status: "needs_review" })); + const progress = canvasAgentProgress(agentRunView({ ...run, snapshot: { ...snapshot, message: statusText }, status: "needs_review" })); writer.write({ type: "data-canvasAgentProgress", id: run.id, data: progress }); await saveMessages({ messages: [{ id: assistantMessageId, chatId: body.id, role: "assistant", parts: [{ type: "text", text }, { type: "data-canvasAgentProgress", data: progress }], attachments: [], createdAt: new Date(), videoStorageStatus: null, temporaryVideoUrl: null, permanentVideoUrl: null, archivedAt: null, videoMetadata: null }] }).catch(() => undefined); - writer.write({ type: "text-delta", id: textId, delta: text }); + emitAssistantText(text.startsWith(streamedText) ? text.slice(streamedText.length) : `\n\n${statusText}`); } finally { clearInterval(heartbeat); writer.write({ type: "text-end", id: textId }); diff --git a/components/canvas/CanvasChatPanel.tsx b/components/canvas/CanvasChatPanel.tsx index 2b815a968..e9f35fc57 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -8932,6 +8932,21 @@ export function CanvasChatPanel({ enabled: effectiveAgentEnabled && !isVideoWorkspace && !isWorkspacePresentation && activeComposerMode === "agent" && isValidChatUuid(rawChatId) && isValidChatUuid(canvasId), chatId: rawChatId, projectId: canvasId, busy: status === "submitted" || status === "streaming", onResults: (projectId, nodeIds) => { void refreshCanvasAfterAgentOperation({ projectId, createdNodeIds: nodeIds }); }, + onRunChange: (run) => { + // The live stream owns its message; polling only recovers disconnected replies. + if (status === "submitted" || status === "streaming") return; + if (!run?.assistantMessageId || !run.assistantText || run.chatId !== activeChatIdRef.current) return; + // Recover the same assistant reply after a stream disconnect or refresh. + setMessages(current => { + const index = current.findIndex(message => message.id === run.assistantMessageId); + if (index < 0) return [...current, { id: run.assistantMessageId!, role: "assistant", parts: [{ type: "text", text: run.assistantText! }] }]; + const oldText = current[index].parts.filter(part => part.type === "text").map(part => part.text).join(""); + if (oldText.startsWith(run.assistantText!)) return current; + return current.map((message, i) => i === index ? { ...message, parts: [ + { type: "text" as const, text: run.assistantText! }, ...message.parts.filter(part => part.type !== "text"), + ] } : message); + }); + }, }); const activeCanvasAgentRun = canvasAgentTask.run; const isCanvasAgentRunning = activeCanvasAgentRun?.status === "running"; diff --git a/components/canvas/LiblibVideoCanvasReplica.tsx b/components/canvas/LiblibVideoCanvasReplica.tsx index e5cc576fd..58f6a79d5 100644 --- a/components/canvas/LiblibVideoCanvasReplica.tsx +++ b/components/canvas/LiblibVideoCanvasReplica.tsx @@ -821,28 +821,28 @@ const STARTER_CARDS = [ title: "故事脚本生成", subtitle: "脚本到分镜", coverUrl: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-story-script-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-story-script-cover.png", steps: ["故事脚本", "多模态文本模型Pro", "可灵 3.0", "预览"], }, { title: "角色三视图", subtitle: "角色资产沉淀", coverUrl: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-character-three-view-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-character-three-view-cover.png", steps: ["角色描述", "角色三视图", "全能图片模型V2", "保存"], }, { title: "全能参考生视频", subtitle: "参考图到成片", coverUrl: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-first-frame-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-first-frame-video-cover.png", steps: ["上传图片", "首帧图", "可灵 3.0", "预览"], }, { title: "音频生视频", subtitle: "音频驱动画面", coverUrl: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-audio-to-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-audio-to-video-cover.png", icon: "music", steps: ["上传音频", "音频主视频", "Wan 2.6", "预览"], }, @@ -865,7 +865,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "预设", category: "周星驰经典名场面", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-first-frame-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-first-frame-video-cover.png", description: "左弧滑行，镜头从人物左侧弧线推进，带轻微喜剧节奏和夸张表情停顿。", duration: "5s", id: "left-arc-slide", @@ -880,7 +880,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "商用", category: "电商展示", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-character-three-view-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-character-three-view-cover.png", description: "电商手机弹出效果，产品从画面中心弹出并旋转定格，适合主图视频开场。", duration: "5s", id: "phone-pop-product", @@ -895,7 +895,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "商用", category: "电商展示", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-story-script-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-story-script-cover.png", description: "咖啡杯出场，杯身从桌面滑入，蒸汽和柔光形成温暖的品牌氛围。", duration: "5s", id: "coffee-cup-entry", @@ -910,7 +910,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "热门", category: "镜头运动", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-audio-to-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-audio-to-video-cover.png", description: "360旋转展示，围绕主体做一圈稳定环绕，适合产品、人物和空间展示。", duration: "6s", id: "orbit-360-showcase", @@ -924,7 +924,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ author: "工业镜头", category: "镜头运动", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/character-image.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/character-image.png", description: "机械臂视角，以第一视角穿过机械臂与产品线，增强科技感和空间纵深。", duration: "5s", id: "robot-arm-pov", @@ -939,7 +939,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "NEW", category: "人物动画", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/character-three-view.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/character-three-view.png", description: "Live 2D，让单张角色图产生眨眼、微笑和轻微身体摆动，适合口播开场。", duration: "4s", id: "live-2d-character", @@ -953,7 +953,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ author: "电影运镜", category: "镜头运动", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-first-frame-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-first-frame-video-cover.png", description: "瞳孔拉近，镜头快速推向眼部并在反光里展示下一场景，适合悬疑转场。", duration: "5s", id: "pupil-zoom", @@ -968,7 +968,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "热门", category: "人物动画", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-story-script-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-story-script-cover.png", description: "飞鸟解体，主体由细碎羽片散开再重组，适合视觉冲击强的短视频片头。", duration: "6s", id: "bird-dissolve", @@ -983,7 +983,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ badge: "商用", category: "电商展示", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-character-three-view-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-character-three-view-cover.png", description: "商品震撼登场，产品带冲击波与局部高光进入画面，适合活动主推款。", duration: "5s", id: "product-hero-entry", @@ -997,7 +997,7 @@ const TOOLBOX_TEMPLATE_PRESETS: ToolboxTemplatePreset[] = [ author: "旅拍模板", category: "镜头运动", cover: - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-audio-to-video-cover.png", + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-audio-to-video-cover.png", description: "旅拍转场，前景遮挡切换到新地点，带自然手持感和轻微速度变化。", duration: "5s", id: "travel-whip-transition", @@ -1160,9 +1160,9 @@ const NINE_GRID_CELL_VISUALS: GridCellVisual[] = [ ]; const TARGET_AVATAR = - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/target-avatar.jpg"; + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/target-avatar.jpg"; const PUBLISH_MODAL_BG_URL = - "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/publish-modal-bg.png"; + "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/publish-modal-bg.png"; const STORY_SCRIPT_PROMPT = "根据我上传的剧本生成一个完整的故事脚本"; const AUDIO_TO_VIDEO_PROMPT = "根据上传的音频生成对应场景画面，镜头语言、节奏、音乐匹配情绪变化，电影级质感。"; @@ -21834,8 +21834,8 @@ const STYLE_MARKET_CATEGORIES = [ "全部", ]; -const STYLE_MARKET_THUMBNAIL_URL = "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-character-three-view-cover.png"; -const EFFECT_MARKET_THUMBNAIL_URL = "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com/video-canvas-demo/starter-first-frame-video-cover.png"; +const STYLE_MARKET_THUMBNAIL_URL = "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-character-three-view-cover.png"; +const EFFECT_MARKET_THUMBNAIL_URL = "https://scqnycdn.jiansun.vip/legacy-oss/demo-sparkai/video-canvas-demo/starter-first-frame-video-cover.png"; const STYLE_MARKET_CARDS: AssetMarketCard[] = [ ["复古马卡龙", "算力潮汐", 177], diff --git a/lib/ai/canvas-agent-runtime/protocol.ts b/lib/ai/canvas-agent-runtime/protocol.ts index 7f8a1f01a..135d0b049 100644 --- a/lib/ai/canvas-agent-runtime/protocol.ts +++ b/lib/ai/canvas-agent-runtime/protocol.ts @@ -38,6 +38,8 @@ export type AgentRunSnapshot = { videoRequirements?: import("./video-protocol").VideoTaskRequirements; objective: string; messages: ModelMessage[]; + /** Public assistant output for this chat reply, retained across disconnects. */ + assistantText?: string; calls: AgentCallRecord[]; inputs: AgentRunInput[]; steps: number; @@ -59,6 +61,7 @@ export type AgentRunSnapshot = { }; export type AgentRunView = { assistantMessageId?: string; + assistantText?: string; activities?: Array<{ id: string; title: string; status: AgentCallRecord["status"] }>; videoSettings?: import("./video-protocol").VideoAgentSettings; id: string; diff --git a/lib/ai/canvas-agent-runtime/store.ts b/lib/ai/canvas-agent-runtime/store.ts index 5f0200202..e07ced104 100644 --- a/lib/ai/canvas-agent-runtime/store.ts +++ b/lib/ai/canvas-agent-runtime/store.ts @@ -37,6 +37,7 @@ export function agentRunView(run: Run): AgentRunView { const stale = run.status === "running" && Date.now() - run.updatedAt.getTime() > LEASE_STALE_MS; return { assistantMessageId: run.assistantMessageId, + assistantText: run.snapshot.assistantText, activities: canvasAgentActivities(run.snapshot), id: run.id, videoSettings: run.snapshot.videoSettings, @@ -151,6 +152,7 @@ export async function acquireAgentRun(params: { if (existing) { if (existing.status === "completed") throw new AgentRunError("该任务已完成，无需重新执行。"); snapshot.message = "正在从已完成的步骤继续"; + snapshot.assistantText = ""; if (snapshot.workspace === "video" && params.snapshot.videoSettings) { snapshot.videoSettings = params.snapshot.videoSettings; snapshot.videoRequirements = { diff --git a/lib/ai/providers/vip98k.ts b/lib/ai/providers/vip98k.ts index b5767d705..6f6c0a543 100644 --- a/lib/ai/providers/vip98k.ts +++ b/lib/ai/providers/vip98k.ts @@ -21,6 +21,7 @@ import { import { getDefaultStorageConfig, getSystemSetting } from "@/lib/db/queries"; import { bindVip98kImages, + bindVip98kExpansionParams, getVip98kImageBindingKeys, hasVerifiedVip98kOpenAiImageArrayContract, normalizeVip98kInvocationHeaders, @@ -1549,6 +1550,9 @@ function buildTaskBody(params: Vip98kTaskParams, invocation: Vip98kInvocation | body.metadata = { ...(toRecord(body.metadata) || {}), mode_image: imageMode }; delete body.mode_image; } + if (params.modelId.toLowerCase() === "image-expansion") { + bindVip98kExpansionParams(body, imageParams); + } const imageBindingKeys = bindVip98kImages(body, params.imageUrls); if (params.imageUrls.length > 0 && !invocation && imageBindingKeys.length === 0) { body.images = params.imageUrls; diff --git a/lib/ai/vip98k-runtime-config.ts b/lib/ai/vip98k-runtime-config.ts index 940cb3077..474c4d3c9 100644 --- a/lib/ai/vip98k-runtime-config.ts +++ b/lib/ai/vip98k-runtime-config.ts @@ -10,9 +10,9 @@ export type Vip98kRuntimeConfig = { }; export const defaultVip98kRuntimeConfig: Vip98kRuntimeConfig = { - submissionTimeoutMs: 120_000, + submissionTimeoutMs: 300_000, modelSubmissionTimeoutMs: { - "nano-banana-pro": 240_000, + "nano-banana-pro": 300_000, }, maxSubmissionAttempts: 3, submissionRetryBaseDelayMs: 5_000, @@ -45,7 +45,7 @@ function normalizeModelTimeouts(value: unknown) { normalized[normalizedModelId] = clampInteger( timeout, defaultVip98kRuntimeConfig.submissionTimeoutMs, - 30_000, + 300_000, 10 * 60_000, ); } @@ -61,7 +61,7 @@ export function normalizeVip98kRuntimeConfig(value: unknown): Vip98kRuntimeConfi submissionTimeoutMs: clampInteger( raw.submissionTimeoutMs, defaultVip98kRuntimeConfig.submissionTimeoutMs, - 30_000, + 300_000, 10 * 60_000, ), modelSubmissionTimeoutMs: normalizeModelTimeouts( diff --git a/lib/ai/vip98k-task-invocation.ts b/lib/ai/vip98k-task-invocation.ts index 6d1a64e60..a6ed58be3 100644 --- a/lib/ai/vip98k-task-invocation.ts +++ b/lib/ai/vip98k-task-invocation.ts @@ -477,3 +477,42 @@ export function resolveVip98kTaskInvocation( return createKnownImageToolInvocation(modelId, knownImageInput); } + +/** Convert final canvas multipliers to the provider's per-side expansion ratios. */ +export function bindVip98kExpansionParams( + body: Record, + params: { xScale?: unknown; yScale?: unknown } +) { + if (params.xScale === undefined && params.yScale === undefined) return; + const ratio = (value: unknown) => { + const scale = Number(value ?? 1.5); + if (!Number.isFinite(scale) || scale <= 1 || scale > 3) + throw new Error("VIP98K 扩图倍数必须大于 1 且不超过 3"); + return Number(((scale - 1) / 2).toFixed(8)); + }; + const directions = { + left: ratio(params.xScale), + right: ratio(params.xScale), + top: ratio(params.yScale), + bottom: ratio(params.yScale), + }; + const metadata = + body.metadata && + typeof body.metadata === "object" && + !Array.isArray(body.metadata) + ? { ...(body.metadata as Record) } + : {}; + const rootDirections = Object.keys(directions).some((key) => key in body); + for (const key of Object.keys(directions)) { + delete body[key]; + delete metadata[key]; + } + if (rootDirections) { + Object.assign(body, directions); + if (Object.keys(metadata).length) body.metadata = metadata; + else delete body.metadata; + } else { + body.metadata = { ...metadata, ...directions }; + } +} + diff --git a/lib/ai/zcode-agent/assistant-output.ts b/lib/ai/zcode-agent/assistant-output.ts new file mode 100644 index 000000000..1353f2c2e --- /dev/null +++ b/lib/ai/zcode-agent/assistant-output.ts @@ -0,0 +1,14 @@ +import type { AgentRunSnapshot } from "../canvas-agent-runtime/protocol"; + +/** Only ordinary assistant text belongs in chat; never use reasoning/tool events. */ +export function publicAssistantText(text: string) { + return text.replace(/[\s\S]*?<\/think>/gi, "") + .replace(/\bZCode\b/gi, "Agent") + .replace(/(?:sk-[\w-]{12,}|Bearer\s+[\w.-]+)/gi, "[已隐藏]"); +} + +export function canvasAgentReplyText(snapshot: Pick) { + const text = snapshot.assistantText || ""; + const final = publicAssistantText(snapshot.message || "").trim(); + return !final || text.trimEnd().endsWith(final) ? text : text ? `${text.trimEnd()}\n\n${final}` : final; +} diff --git a/lib/ai/zcode-agent/model-gateway.ts b/lib/ai/zcode-agent/model-gateway.ts index 06a1ab18b..089d9ddde 100644 --- a/lib/ai/zcode-agent/model-gateway.ts +++ b/lib/ai/zcode-agent/model-gateway.ts @@ -70,12 +70,15 @@ export async function createZcodeModelGateway(configured: ConfiguredModel, onErr attemptSignal.throwIfAborted(); const request = target.apiType === "openai-responses" ? toResponsesRequest(body) : { ...body }; delete request.stream_options; - if (isDeepSeekOfficialUrl(target.endpoint.href)) { - // The native core explicitly sends thinking=enabled, overriding - // the provider default. Its history omits reasoning_content, so - // enforce the compatible tool protocol at this Agent boundary. + // Agent runs use ordinary replies, including on fallback channels. + delete request.thinking; + delete request.enable_thinking; + delete request.reasoning_effort; + delete request.reasoning; + if (isDeepSeekOfficialUrl(target.endpoint.href) || /deepseek|kimi|glm/i.test(target.modelId)) { request.thinking = { type: "disabled" }; - delete request.reasoning_effort; + } else if (/qwen/i.test(target.modelId) || ["alibaba", "qwen", "local-qwen"].includes(entry.provider.type)) { + request.enable_thinking = false; } const providerFetch = isDeepSeekOfficialUrl(target.endpoint.href) ? createDeepSeekCompatibleFetch() : fetch; const response = await providerFetch(target.endpoint, { method: "POST", redirect: "error", diff --git a/lib/ai/zcode-agent/task.ts b/lib/ai/zcode-agent/task.ts index 9fa0f35de..06250f1c4 100644 --- a/lib/ai/zcode-agent/task.ts +++ b/lib/ai/zcode-agent/task.ts @@ -8,6 +8,7 @@ import { AgentToolRejectedError, publicAgentError, type AgentRunSnapshot, type C import { isCompletedSimpleCutout } from "@/lib/ai/canvas-agent-runtime/cutout-completion"; import { requiresCanvasInteraction } from "@/lib/ai/canvas-agent-runtime/results"; import { AgentImageGenerationError } from "@/lib/ai/runtime/image-generation-error"; +import { publicAssistantText } from "./assistant-output"; type Ports = { native?: typeof runNativeZcode; @@ -19,6 +20,7 @@ type Ports = { context: unknown; gateway: { modelId: string; apiType: string; forward: (body: Record, path: string, signal: AbortSignal) => Promise }; signal: AbortSignal; + onAssistantText?: (delta: string) => void; checkpoint: (snapshot: AgentRunSnapshot, status: CanvasAgentRunStatus) => Promise; consumeInputs: (snapshot: AgentRunSnapshot) => Promise<{ cancelled: boolean }>; isCancelled: () => Promise; @@ -39,6 +41,15 @@ export async function runZcodeCanvasTask(ports: Ports): Promise(); + let lastTextMessage = ""; + async function appendText(delta: string) { + delta = publicAssistantText(delta).slice(0, Math.max(0, 100_000 - (snapshot.assistantText?.length || 0))); + if (!delta) return; + snapshot.assistantText = (snapshot.assistantText || "") + delta; + await ports.checkpoint(structuredClone(snapshot), "running"); + ports.onAssistantText?.(delta); + } async function finishVerifiedCutout() { snapshot.nativeRuntime!.state = "stopped"; snapshot.messages.push({ role: "assistant", content: verifiedCompletion }); @@ -100,6 +111,18 @@ export async function runZcodeCanvasTask(ports: Ports): Promise { + if (event.type === "model.streaming" && event.payload?.kind === "text_delta" && typeof event.payload.delta === "string") { + // Child sessions and reasoning/tool-argument deltas are private. + if (snapshot.nativeRuntime?.sessionId && event.sessionId !== snapshot.nativeRuntime.sessionId) return; + await exclusive(async () => { + if (event.eventId && seenTextEvents.has(event.eventId)) return; + if (event.eventId) seenTextEvents.add(event.eventId); + const messageId = String(event.payload!.assistantMessageId || ""); + await appendText((lastTextMessage && lastTextMessage !== messageId ? "\n\n" : "") + event.payload!.delta); + lastTextMessage = messageId; + }); + return; + } if (!["turn.started", "result", "session.created", "session.resumed"].includes(event.type)) return; await exclusive(async () => { if (event.sessionId) snapshot.nativeRuntime!.sessionId = event.sessionId; @@ -196,6 +219,9 @@ export async function runZcodeCanvasTask(ports: Ports): Promise /// -import "./.next/types/routes.d.ts"; +import "./.next-agent-brain-icon-20261022/types/routes.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/app/api-reference/config/typescript for more information. diff --git a/scripts/storage/audit-aliyun-qiniu-references.cjs b/scripts/storage/audit-aliyun-qiniu-references.cjs new file mode 100644 index 000000000..779e7a3a2 --- /dev/null +++ b/scripts/storage/audit-aliyun-qiniu-references.cjs @@ -0,0 +1,6 @@ +/* Read-only audit of live business references; obsolete admin configuration backups stay intact. */ +const fs=require('fs'),path=require('path');const root=process.env.MIGRATION_APP_ROOT||process.cwd(),dir=process.env.MIGRATION_RUN_DIR||'/var/lib/ai-chatbot/aliyun-qiniu-20261014';const {urlsIn}=require('./migrate-aliyun-to-qiniu.cjs');require(path.join(root,'node_modules/dotenv')).config({path:path.join(root,'.env.local')});const sql=require(path.join(root,'node_modules/postgres'))(process.env.POSTGRES_URL,{max:1,prepare:false,connection:{statement_timeout:600000}});const host='demo-sparkai.oss-cn-shanghai.aliyuncs.com',quote=s=>'"'+s.replaceAll('"','""')+'"'; +(async()=>{try{const verified=new Set(fs.readFileSync(path.join(dir,'verified.jsonl'),'utf8').split('\n').filter(Boolean).map(s=>JSON.parse(s).id)),unavailable=new Set(fs.existsSync(path.join(dir,'unavailable-source.json'))?JSON.parse(fs.readFileSync(path.join(dir,'unavailable-source.json'))).map(r=>r.id):[]),remaining=new Map(),tables=new Map(),excluded=new Set(['StorageConfig','StorageMigrationTask','ImageArchiveJob','__drizzle_migrations','SystemSettingHistory','SystemSettingSnapshot']); +const columns=await sql`select c.table_name as "tableName",c.column_name as "columnName" from information_schema.columns c join information_schema.tables t on c.table_schema=t.table_schema and c.table_name=t.table_name where c.table_schema='public' and t.table_type='BASE TABLE' and c.data_type in ('text','character varying','json','jsonb') order by c.table_name,c.ordinal_position`; +for(const c of columns)if(!excluded.has(c.tableName)){if(!tables.has(c.tableName))tables.set(c.tableName,[]);tables.get(c.tableName).push(c.columnName);}const tableCounts={};for(const [table,cols] of tables){const where=cols.map(c=>'('+quote(c)+'::text like $1'+(table==='ImageArchiveTask'&&c==='sourceUrl'?" and coalesce(metadata->>'legacyOssMigrationRun','') <> '20261014'":'')+')').join(' or ');let matched=0;for await(const rows of sql.unsafe('select '+cols.map(quote).join(',')+' from '+quote(table)+' where '+where,['%'+host+'/%']).cursor(50))for(const row of rows){matched++;for(const [column,value]of Object.entries(row)){if(table==='ImageArchiveTask'&&column==='sourceUrl'&&row.metadata?.legacyOssMigrationRun==='20261014')continue;for(const r of urlsIn(value))if(r.sourceHost===host){if(!remaining.has(r.id))remaining.set(r.id,{...r,verified:verified.has(r.id),knownUnavailable:unavailable.has(r.id),locations:{}});const item=remaining.get(r.id),key=table+'.'+column;item.locations[key]=(item.locations[key]||0)+1;}}}if(matched){tableCounts[table]=matched;console.log('remaining_reference_rows',JSON.stringify({table,rows:matched}));}} +const resources=[...remaining.values()],summary={auditedAt:new Date().toISOString(),remainingUniqueResources:resources.length,verifiedPendingRewrite:resources.filter(r=>r.verified).length,knownUnavailable:resources.filter(r=>r.knownUnavailable&&!r.verified).length,unclassified:resources.filter(r=>!r.knownUnavailable&&!r.verified).length,tableCounts};fs.writeFileSync(path.join(dir,'reference-audit-resources.json'),JSON.stringify(resources),{mode:0o600});fs.writeFileSync(path.join(dir,'reference-audit-summary.json'),JSON.stringify(summary,null,2),{mode:0o600});console.log('reference_audit',JSON.stringify(summary));}finally{await sql.end()}})().catch(e=>{console.error('audit_failed',e.name,String(e.message).replace(/https?:\/\/\S+/g,'[URL]').slice(0,180));process.exitCode=1}); diff --git a/scripts/storage/inventory-aliyun-qiniu-source.cjs b/scripts/storage/inventory-aliyun-qiniu-source.cjs new file mode 100644 index 000000000..9820f8d3d --- /dev/null +++ b/scripts/storage/inventory-aliyun-qiniu-source.cjs @@ -0,0 +1,8 @@ +const fs=require('fs'),path=require('path'),root=process.env.MIGRATION_APP_ROOT||process.cwd(),dir=process.env.MIGRATION_RUN_DIR||'/var/lib/ai-chatbot/aliyun-qiniu-20261014'; +require(path.join(root,'node_modules/dotenv')).config({path:path.join(root,'.env.local')});const sql=require(path.join(root,'node_modules/postgres'))(process.env.POSTGRES_URL,{max:1,prepare:false});const {S3Client,ListObjectsV2Command}=require(path.join(root,'node_modules/@aws-sdk/client-s3')); +(async()=>{let client;try{const complete=path.join(dir,'source-metadata.complete');if(fs.existsSync(complete)){console.log('Source metadata inventory already completed');return;}const plan=JSON.parse(fs.readFileSync(path.join(dir,'plan-all.json'))),wanted=new Set(plan.filter(r=>!r.variant).map(r=>r.sourceKey)); +const [s]=await sql`select * from "StorageConfig" where bucket='demo-sparkai' limit 1`;client=new S3Client({endpoint:s.endpoint,region:s.region,forcePathStyle:false,credentials:{accessKeyId:s.accessKeyId,secretAccessKey:s.secretAccessKey},maxAttempts:3});let marker,listed=0,matched=0;const checkpointFile=path.join(dir,'source-list-checkpoint.json'),file=path.join(dir,'source-metadata.jsonl');if(fs.existsSync(checkpointFile)){const c=JSON.parse(fs.readFileSync(checkpointFile));marker=c.marker;listed=c.listed;matched=c.matched;} +for(;;){const r=await client.send(new ListObjectsV2Command({Bucket:s.bucket,MaxKeys:1000,ContinuationToken:marker}),{abortSignal:AbortSignal.timeout(60000)});const lines=[];for(const o of r.Contents||[]){listed++;if(wanted.has(o.Key)){lines.push(JSON.stringify({key:o.Key,size:o.Size,md5:String(o.ETag||'').replaceAll('"','').toLowerCase()}));matched++;}} +if(lines.length)fs.appendFileSync(file,lines.join('\n')+'\n',{mode:0o600});marker=r.NextContinuationToken;fs.writeFileSync(checkpointFile,JSON.stringify({marker,listed,matched}),{mode:0o600});if(listed%25000<1000)console.log(JSON.stringify({listed,matched,wanted:wanted.size}));if(!r.IsTruncated)break;} +fs.writeFileSync(complete,JSON.stringify({listed,matched,wanted:wanted.size,completedAt:new Date().toISOString()}),{mode:0o600});console.log('Source inventory complete',JSON.stringify({listed,matched,wanted:wanted.size})); +}finally{client?.destroy();await sql.end()}})().catch(e=>{console.error('source_inventory_failed',e.name,String(e.message).replace(/https?:\/\/\S+/g,'[URL]').slice(0,180));process.exitCode=1}); diff --git a/scripts/storage/migrate-aliyun-to-minio.ts b/scripts/storage/migrate-aliyun-to-minio.ts new file mode 100644 index 000000000..41903d445 --- /dev/null +++ b/scripts/storage/migrate-aliyun-to-minio.ts @@ -0,0 +1,1308 @@ +import { + DeleteObjectCommand, + GetObjectCommand, + HeadObjectCommand, + ListObjectsV2Command, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3"; +import { createHash } from "node:crypto"; +import { Transform, type Readable } from "node:stream"; +import postgres from "postgres"; + +type StorageConfig = { + id: string; + name: string; + type: string; + endpoint: string | null; + region: string | null; + bucket: string | null; + accessKeyId: string | null; + secretAccessKey: string | null; + publicUrl: string | null; + isEnabled: boolean; +}; + +type MigrationTask = { + id: string; + sourceUrl: string; + sourcePath: string; + attemptCount: number; +}; + +const SOURCE_PREFIX = "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com"; +const PROJECT_IMAGE_EXTENSION_PATTERN = + "(?:png|jpe?g|webp|jfif|gif|avif|tiff?|heic|heif|bmp|svg)"; +const PROJECT_IMAGE_PATH_SQL_PATTERN = + `^uploads/[A-Za-z0-9._/-]+\\.${PROJECT_IMAGE_EXTENSION_PATTERN}$`; +const PROJECT_IMAGE_SOURCE_URL_PATTERN = + `(https://demo-sparkai\\.oss-cn-shanghai\\.aliyuncs\\.com/${PROJECT_IMAGE_PATH_SQL_PATTERN.slice(1, -1)})`; +const PROJECT_IMAGE_SOURCE_PATH_CAPTURE_PATTERN = + `https://demo-sparkai\\.oss-cn-shanghai\\.aliyuncs\\.com/(${PROJECT_IMAGE_PATH_SQL_PATTERN.slice(1, -1)})`; +const PROJECT_IMAGE_PATH_PATTERN = new RegExp( + PROJECT_IMAGE_PATH_SQL_PATTERN, + "i" +); +const TASK_STATUSES = ["pending", "failed", "copying", "copied"]; +const args = process.argv.slice(2); +const command = args[0] || "audit"; + +function readArg(name: string, fallback: string) { + const prefix = `--${name}=`; + return args.find((item) => item.startsWith(prefix))?.slice(prefix.length) || fallback; +} + +function readIntArg(name: string, fallback: number, min = 0, max = Number.MAX_SAFE_INTEGER) { + const parsed = Number(readArg(name, String(fallback))); + if (!Number.isFinite(parsed)) return fallback; + return Math.max(min, Math.min(max, Math.round(parsed))); +} + +function cleanKey(value: string) { + return String(value || "").replace(/^\/+/, ""); +} + +function cleanUrl(value: string | null | undefined) { + return String(value || "").trim().replace(/\/+$/, ""); +} + +function formatBytes(value: number) { + const units = ["B", "KiB", "MiB", "GiB", "TiB"]; + let amount = value; + let index = 0; + while (amount >= 1024 && index < units.length - 1) { + amount /= 1024; + index += 1; + } + return `${amount.toFixed(index === 0 ? 0 : 2)} ${units[index]}`; +} + +function errorMessage(error: unknown) { + const record = (error && typeof error === "object" ? error : {}) as { + name?: string; + message?: string; + code?: string; + Code?: string; + cause?: unknown; + $metadata?: { + httpStatusCode?: number; + requestId?: string; + extendedRequestId?: string; + attempts?: number; + }; + }; + const cause = + record.cause && record.cause !== error + ? String( + record.cause instanceof Error + ? `${record.cause.name}: ${record.cause.message}` + : record.cause + ) + : undefined; + return JSON.stringify({ + name: record.name || (error instanceof Error ? error.name : undefined), + message: record.message || (error instanceof Error ? error.message : String(error || "Unknown error")), + code: record.code || record.Code, + status: record.$metadata?.httpStatusCode, + requestId: record.$metadata?.requestId, + extendedRequestId: record.$metadata?.extendedRequestId, + attempts: record.$metadata?.attempts, + cause, + }).slice(0, 1800); +} + +function normalizeEtag(value: string | undefined) { + return String(value || "").replace(/^"|"$/g, ""); +} + +function isSimpleEtag(value: string | undefined) { + const normalized = normalizeEtag(value); + return /^[a-f0-9]{32}$/i.test(normalized); +} + +function assertStorageConfig(config: StorageConfig | undefined, role: string) { + if ( + !config || + !config.endpoint || + !config.bucket || + !config.accessKeyId || + !config.secretAccessKey + ) { + throw new Error(`${role} storage configuration is incomplete`); + } + return config as StorageConfig & { + endpoint: string; + bucket: string; + accessKeyId: string; + secretAccessKey: string; + }; +} + +function createClient(config: ReturnType) { + const endpoint = cleanUrl(config.endpoint); + const isCloud = /aliyuncs\.com|amazonaws\.com/i.test(endpoint); + return new S3Client({ + endpoint, + region: config.region || "auto", + credentials: { + accessKeyId: config.accessKeyId, + secretAccessKey: config.secretAccessKey, + }, + forcePathStyle: config.type === "qiniu-kodo" ? true : !isCloud, + maxAttempts: 4, + }); +} + +async function retry(operation: () => Promise, attempts = 3) { + let lastError: unknown; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + try { + return await operation(); + } catch (error) { + lastError = error; + if (attempt < attempts) { + await new Promise((resolve) => setTimeout(resolve, attempt * 1500)); + } + } + } + throw lastError; +} + +async function loadStorageConfigs(sql: ReturnType) { + const settings = await sql<{ value: Record }[]>` + select "value" + from "SystemSettings" + where "key" = 'oss_image_migration_config' + limit 1 + `; + const value = settings[0]?.value || {}; + const sourceId = String(value.sourceStorageConfigId || ""); + const targetId = String(value.targetStorageConfigId || ""); + if (!sourceId || !targetId) { + throw new Error("Migration source or target storage ID is not configured"); + } + + const configs = await sql` + select + "id", "name", "type", "endpoint", "region", "bucket", + "accessKeyId", "secretAccessKey", "publicUrl", "isEnabled" + from "StorageConfig" + where "id" in (${sourceId}, ${targetId}) + `; + const source = assertStorageConfig(configs.find((item) => item.id === sourceId), "Source"); + const target = assertStorageConfig(configs.find((item) => item.id === targetId), "Target"); + if (!/aliyuncs\.com/i.test(source.endpoint)) { + throw new Error(`Refusing to run: source is not Aliyun OSS (${source.name})`); + } + if (/aliyuncs\.com/i.test(target.endpoint)) { + throw new Error(`Refusing to run: target is also Aliyun OSS (${target.name})`); + } + if (!cleanUrl(target.publicUrl)) { + throw new Error("Target storage public URL is missing"); + } + return { source, target }; +} + +function targetUrl(config: StorageConfig, key: string) { + return `${cleanUrl(config.publicUrl)}/${cleanKey(key)}`; +} + +async function headPublicObject(url: string) { + const response = await fetch(url, { + method: "HEAD", + headers: { + "Accept-Encoding": "identity", + "User-Agent": "AI-Chatbot-Storage-Migrator/1.0", + }, + signal: AbortSignal.timeout(20_000), + }); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`Target public HEAD failed: HTTP ${response.status} (${url})`); + } + let contentLength = Number(response.headers.get("content-length") || 0); + let etag = response.headers.get("etag") || undefined; + let contentType = response.headers.get("content-type") || undefined; + if (contentLength <= 0) { + const rangeResponse = await fetch(url, { + headers: { + "Accept-Encoding": "identity", + Range: "bytes=0-0", + "User-Agent": "AI-Chatbot-Storage-Migrator/1.0", + }, + signal: AbortSignal.timeout(20_000), + }); + if (!rangeResponse.ok) { + throw new Error(`Target public range read failed: HTTP ${rangeResponse.status} (${url})`); + } + const contentRange = rangeResponse.headers.get("content-range") || ""; + contentLength = Number( + contentRange.match(/\/(\d+)$/)?.[1] || + rangeResponse.headers.get("content-length") || + 0 + ); + etag = etag || rangeResponse.headers.get("etag") || undefined; + contentType = contentType || rangeResponse.headers.get("content-type") || undefined; + await rangeResponse.body?.cancel().catch(() => undefined); + } + return { + contentLength, + etag, + contentType, + }; +} + +async function copyObject(params: { + sourceClient: S3Client; + targetClient: S3Client; + source: ReturnType; + target: ReturnType; + key: string; +}) { + const key = cleanKey(params.key); + const sourceHead = await retry(() => + params.sourceClient.send(new HeadObjectCommand({ Bucket: params.source.bucket, Key: key })) + ); + const sourceBytes = Number(sourceHead.ContentLength || 0); + if (sourceBytes <= 0) throw new Error(`Source object is empty: ${key}`); + if (sourceBytes > 5 * 1024 * 1024 * 1024) { + throw new Error(`Object exceeds single-upload limit: ${key} (${sourceBytes})`); + } + + const publicUrl = targetUrl(params.target, key); + let transferError: unknown; + for (let attempt = 1; attempt <= 3; attempt += 1) { + try { + const sourceObject = await params.sourceClient.send( + new GetObjectCommand({ Bucket: params.source.bucket, Key: key }) + ); + if (!sourceObject.Body) throw new Error(`Source object body is missing: ${key}`); + + const body = sourceObject.Body as Readable; + if (typeof body.pipe !== "function") { + throw new Error(`Source object body is not a Node.js stream: ${key}`); + } + const sourceHash = createHash("md5"); + const hashingStream = new Transform({ + transform(chunk: Buffer, _encoding, callback) { + sourceHash.update(chunk); + callback(null, chunk); + }, + }); + const uploadResult = await params.targetClient.send( + new PutObjectCommand({ + Bucket: params.target.bucket, + Key: key, + Body: body.pipe(hashingStream) as never, + ContentLength: sourceBytes, + ContentType: sourceHead.ContentType || "application/octet-stream", + CacheControl: sourceHead.CacheControl || "public, max-age=31536000, immutable", + ContentDisposition: sourceHead.ContentDisposition, + Metadata: sourceHead.Metadata, + ACL: "public-read", + }) + ); + const sourceMd5 = sourceHash.digest("hex"); + + const copiedHead = await retry(() => headPublicObject(publicUrl), 5); + if (!copiedHead || copiedHead.contentLength !== sourceBytes) { + throw new Error( + `Target size mismatch: ${key} source=${sourceBytes} target=${copiedHead?.contentLength}` + ); + } + if ( + isSimpleEtag(uploadResult.ETag || copiedHead.etag) && + sourceMd5 !== normalizeEtag(uploadResult.ETag || copiedHead.etag) + ) { + throw new Error(`Target checksum mismatch: ${key}`); + } + return { bytes: sourceBytes, reused: false }; + } catch (error) { + transferError = error; + if (attempt < 3) { + await new Promise((resolve) => setTimeout(resolve, attempt * 2000)); + } + } + } + throw transferError; +} + +async function runPool(items: T[], concurrency: number, worker: (item: T) => Promise) { + let cursor = 0; + await Promise.all( + Array.from({ length: Math.min(concurrency, items.length) }, async () => { + while (cursor < items.length) { + const item = items[cursor]; + cursor += 1; + await worker(item); + } + }) + ); +} + +async function audit(sql: ReturnType, source: ReturnType, target: ReturnType) { + const statuses = await sql<{ status: string; count: number }[]>` + select "status", count(*)::int as "count" + from "StorageMigrationTask" + where "sourceStorageConfigId" = ${source.id} + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + group by "status" + order by "status" + `; + console.log("storage", { + source: { name: source.name, bucket: source.bucket, enabled: source.isEnabled }, + target: { name: target.name, bucket: target.bucket, publicUrl: cleanUrl(target.publicUrl) }, + }); + console.log("task_statuses", statuses); + + if (readArg("list", "false") !== "true") return; + + const taskRows = await sql<{ sourcePath: string }[]>` + select "sourcePath" + from "StorageMigrationTask" + where "status" in ${sql(TASK_STATUSES)} + and "sourceStorageConfigId" = ${source.id} + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + `; + const taskKeys = new Set(taskRows.map((item) => cleanKey(item.sourcePath))); + const sourceClient = createClient(source); + const targetClient = createClient(target); + const sourceSizes = new Map(); + + async function list(client: S3Client, bucket: string, label: string) { + let continuationToken: string | undefined; + let count = 0; + let bytes = 0; + let taskCount = 0; + let taskBytes = 0; + let maxBytes = 0; + do { + const page = await client.send( + new ListObjectsV2Command({ + Bucket: bucket, + Prefix: "uploads/", + ContinuationToken: continuationToken, + }) + ); + for (const item of page.Contents || []) { + const key = cleanKey(item.Key || ""); + const size = Number(item.Size || 0); + count += 1; + bytes += size; + maxBytes = Math.max(maxBytes, size); + if (taskKeys.has(key)) { + taskCount += 1; + taskBytes += size; + } + if (label === "source") sourceSizes.set(key, size); + } + continuationToken = page.IsTruncated ? page.NextContinuationToken : undefined; + if (count > 0 && count % 100_000 < 1000) { + console.log(`${label}_list_progress`, { count, bytes: formatBytes(bytes) }); + } + } while (continuationToken); + console.log(`${label}_inventory`, { + count, + bytes: formatBytes(bytes), + taskCount, + taskBytes: formatBytes(taskBytes), + extraObjects: count - taskCount, + missingTaskObjects: taskKeys.size - taskCount, + maxObject: formatBytes(maxBytes), + }); + } + + try { + await list(sourceClient, source.bucket, "source"); + await list(targetClient, target.bucket, "target"); + } finally { + sourceClient.destroy(); + targetClient.destroy(); + sourceSizes.clear(); + } +} + +async function probe(sql: ReturnType, source: ReturnType, target: ReturnType) { + const rows = await sql` + select "id", "sourceUrl", "sourcePath", "attemptCount" + from "StorageMigrationTask" + where "status" in ('pending', 'failed') + and "sourceStorageConfigId" = ${source.id} + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" not like '%..%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and "sourceUrl" = ${`${SOURCE_PREFIX}/`} || "sourcePath" + order by "createdAt" desc + limit 20 + `; + if (rows.length === 0) throw new Error("No pending migration task is available for probe"); + + const sourceClient = createClient(source); + const targetClient = createClient(target); + const task = rows[0]; + const originalKey = cleanKey(task.sourcePath); + const probeKey = `migration-probe/${Date.now()}-${originalKey.split("/").pop()}`; + try { + console.log("probe_stage", { stage: "source_head", key: originalKey }); + const sourceHead = await sourceClient.send( + new HeadObjectCommand({ Bucket: source.bucket, Key: originalKey }) + ); + console.log("probe_stage", { stage: "source_get", bytes: sourceHead.ContentLength }); + const sourceObject = await sourceClient.send( + new GetObjectCommand({ Bucket: source.bucket, Key: originalKey }) + ); + if (!sourceObject.Body) throw new Error("Probe source body is missing"); + console.log("probe_stage", { stage: "target_put", key: probeKey }); + await targetClient.send( + new PutObjectCommand({ + Bucket: target.bucket, + Key: probeKey, + Body: sourceObject.Body as never, + ContentLength: sourceHead.ContentLength, + ContentType: sourceHead.ContentType || "application/octet-stream", + CacheControl: "no-store", + ACL: "public-read", + }) + ); + const publicUrl = targetUrl(target, probeKey); + console.log("probe_stage", { stage: "target_public_read" }); + const response = await fetch(publicUrl, { + headers: { Range: "bytes=0-0", "User-Agent": "AI-Chatbot-Storage-Migration-Probe/1.0" }, + signal: AbortSignal.timeout(20_000), + }); + if (!response.ok) { + throw new Error(`Probe public read failed: HTTP ${response.status}`); + } + const contentRange = response.headers.get("content-range") || ""; + const responseBytes = Number(contentRange.match(/\/(\d+)$/)?.[1] || 0); + if (responseBytes > 0 && responseBytes !== Number(sourceHead.ContentLength || 0)) { + throw new Error( + `Probe public size mismatch: source=${sourceHead.ContentLength} target=${responseBytes}` + ); + } + console.log("probe_ok", { + sourceKey: originalKey, + bytes: sourceHead.ContentLength, + contentType: sourceHead.ContentType, + targetPublicReadStatus: response.status, + }); + } finally { + await targetClient + .send(new DeleteObjectCommand({ Bucket: target.bucket, Key: probeKey })) + .catch(() => undefined); + sourceClient.destroy(); + targetClient.destroy(); + } +} + +async function copyTasks(sql: ReturnType, source: ReturnType, target: ReturnType) { + const concurrency = readIntArg("concurrency", 4, 1, 16); + const batchSize = readIntArg("batch", concurrency * 5, concurrency, 500); + const limit = readIntArg("limit", 100, 0); + const maxAttempts = readIntArg("max-attempts", 12, 1, 100); + let stopping = false; + let processed = 0; + let succeeded = 0; + let failed = 0; + let bytes = 0; + let reused = 0; + const startedAt = Date.now(); + process.once("SIGINT", () => { + stopping = true; + }); + process.once("SIGTERM", () => { + stopping = true; + }); + + await sql` + update "StorageMigrationTask" + set "status" = 'failed', + "lastError" = 'Recovered stale copying task', + "updatedAt" = now() + where "status" = 'copying' + and "sourceStorageConfigId" = ${source.id} + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and ( + "targetStorageConfigId" is null + or "targetStorageConfigId" = ${target.id} + ) + and "updatedAt" < now() - interval '30 minutes' + `; + + const sourceClient = createClient(source); + const targetClient = createClient(target); + try { + while (!stopping && (limit === 0 || processed < limit)) { + const remaining = limit === 0 ? batchSize : Math.min(batchSize, limit - processed); + const tasks = await sql.begin(async (transaction) => { + return transaction` + with selected as ( + select "id" + from "StorageMigrationTask" + where ( + "status" = 'pending' + or ( + "status" = 'failed' + and "updatedAt" < now() - interval '10 minutes' + ) + ) + and "sourceStorageConfigId" = ${source.id} + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" not like '%..%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and ( + "targetStorageConfigId" is null + or "targetStorageConfigId" = ${target.id} + ) + and "attemptCount" < ${maxAttempts} + order by "createdAt", "id" + for update skip locked + limit ${remaining} + ) + update "StorageMigrationTask" task + set "status" = 'copying', + "attemptCount" = task."attemptCount" + 1, + "lastError" = null, + "updatedAt" = now() + from selected + where task."id" = selected."id" + returning task."id", task."sourceUrl", task."sourcePath", task."attemptCount" + `; + }); + if (tasks.length === 0) break; + + await runPool(tasks, concurrency, async (task) => { + try { + if (!task.sourceUrl.startsWith(`${SOURCE_PREFIX}/uploads/`)) { + throw new Error(`Unexpected source URL: ${task.sourceUrl}`); + } + if ( + !task.sourcePath.startsWith("uploads/") || + task.sourcePath.includes("..") || + !PROJECT_IMAGE_PATH_PATTERN.test(task.sourcePath) || + task.sourceUrl !== `${SOURCE_PREFIX}/${task.sourcePath}` + ) { + throw new Error( + `Project object scope mismatch: ${task.sourceUrl} (${task.sourcePath})` + ); + } + const result = await copyObject({ + sourceClient, + targetClient, + source, + target, + key: task.sourcePath, + }); + const url = targetUrl(target, task.sourcePath); + await sql` + update "StorageMigrationTask" + set "status" = 'copied', + "targetUrl" = ${url}, + "targetStorageConfigId" = ${target.id}, + "lastError" = null, + "updatedAt" = now() + where "id" = ${task.id} + `; + succeeded += 1; + bytes += result.bytes; + if (result.reused) reused += 1; + } catch (error) { + failed += 1; + await sql` + update "StorageMigrationTask" + set "status" = 'failed', + "lastError" = ${errorMessage(error)}, + "updatedAt" = now() + where "id" = ${task.id} + `; + } finally { + processed += 1; + if (processed % 25 === 0 || processed === tasks.length) { + const seconds = Math.max(1, (Date.now() - startedAt) / 1000); + console.log("copy_progress", { + processed, + succeeded, + failed, + reused, + bytes: formatBytes(bytes), + objectsPerSecond: Number((processed / seconds).toFixed(2)), + }); + } + } + }); + } + } finally { + sourceClient.destroy(); + targetClient.destroy(); + } + console.log("copy_finished", { + processed, + succeeded, + failed, + reused, + bytes: formatBytes(bytes), + stopped: stopping, + }); +} + +function quoteIdentifier(value: string) { + return `"${value.replaceAll('"', '""')}"`; +} + +async function inventoryDatabaseReferences(sql: ReturnType) { + const sourcePrefix = `${SOURCE_PREFIX}/uploads/`; + const excludedTables = new Set([ + "StorageConfig", + "StorageMigrationTask", + "SystemSettings", + "SystemSettingHistory", + "SystemSettingSnapshot", + "__drizzle_migrations", + ]); + const columns = await sql< + { tableName: string; columnName: string; dataType: string }[] + >` + select + table_name as "tableName", + column_name as "columnName", + data_type as "dataType" + from information_schema.columns + where table_schema = 'public' + and data_type in ('text', 'character varying', 'json', 'jsonb') + order by table_name, ordinal_position + `; + const candidates = columns.filter((item) => !excludedTables.has(item.tableName)); + const matches: Array<{ + tableName: string; + columnName: string; + dataType: string; + count: number; + }> = []; + const timeouts: Array<{ tableName: string; columnName: string; error: string }> = []; + + for (let index = 0; index < candidates.length; index += 1) { + const item = candidates[index]; + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + try { + const rows = await sql.unsafe<{ count: string }[]>( + `select count(*)::text as count from ${table} where ${column}::text ~* $1`, + [PROJECT_IMAGE_SOURCE_URL_PATTERN] + ); + const count = Number(rows[0]?.count || 0); + if (count > 0) { + const match = { ...item, count }; + matches.push(match); + console.log("reference_column", match); + } + } catch (error) { + const timeout = { + tableName: item.tableName, + columnName: item.columnName, + error: errorMessage(error), + }; + timeouts.push(timeout); + console.log("reference_scan_failed", timeout); + } + if ((index + 1) % 25 === 0) { + console.log("reference_scan_progress", { + scanned: index + 1, + total: candidates.length, + matchedColumns: matches.length, + failedColumns: timeouts.length, + }); + } + } + console.log("reference_inventory", { + sourcePrefix, + scannedColumns: candidates.length, + matchedColumns: matches.length, + failedColumns: timeouts.length, + totalMatchingRowsAcrossColumns: matches.reduce((sum, item) => sum + item.count, 0), + matches, + timeouts, + }); +} + +async function auditReferenceCoverage( + sql: ReturnType, + source: ReturnType, + target: ReturnType +) { + const sourcePrefix = `${SOURCE_PREFIX}/uploads/`; + const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN; + const excludedTables = new Set([ + "StorageConfig", + "StorageMigrationTask", + "SystemSettings", + "SystemSettingHistory", + "SystemSettingSnapshot", + "__drizzle_migrations", + ]); + const columns = await sql< + { tableName: string; columnName: string; dataType: string }[] + >` + select + table_name as "tableName", + column_name as "columnName", + data_type as "dataType" + from information_schema.columns + where table_schema = 'public' + and data_type in ('text', 'character varying', 'json', 'jsonb') + order by table_name, ordinal_position + `; + const candidates = columns.filter((item) => !excludedTables.has(item.tableName)); + + await sql.begin(async (transaction) => { + await transaction.unsafe( + "create temporary table migration_referenced_url (url text primary key) on commit drop" + ); + for (let index = 0; index < candidates.length; index += 1) { + const item = candidates[index]; + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + await transaction.unsafe( + ` + insert into migration_referenced_url (url) + select distinct (matches_array)[1] + from ${table} + cross join lateral regexp_matches(${column}::text, $1, 'gi') + as extracted(matches_array) + where ${column}::text like $2 + on conflict (url) do nothing + `, + [sourcePattern, `%${sourcePrefix}%`] + ); + if ((index + 1) % 25 === 0) { + const progress = await transaction<{ count: string }[]>` + select count(*)::text as "count" from migration_referenced_url + `; + console.log("reference_coverage_progress", { + scanned: index + 1, + total: candidates.length, + distinctUrls: Number(progress[0]?.count || 0), + }); + } + } + + const coverage = await transaction< + { + distinctReferencedUrls: string; + coveredByTask: string; + missingTask: string; + copied: string; + unresolved: string; + }[] + >` + select + count(*)::text as "distinctReferencedUrls", + count(task."id")::text as "coveredByTask", + count(*) filter (where task."id" is null)::text as "missingTask", + count(*) filter ( + where task."status" in ('copied', 'migrated', 'deleted') + and task."targetStorageConfigId" = ${target.id} + and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + )::text as "copied", + count(*) filter ( + where task."id" is null + or task."status" not in ('copied', 'migrated', 'deleted') + or task."targetStorageConfigId" is distinct from ${target.id} + or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + )::text as "unresolved" + from migration_referenced_url referenced + left join "StorageMigrationTask" task + on task."sourceUrl" = referenced.url + and task."sourceStorageConfigId" = ${source.id} + `; + const missingSamples = await transaction<{ url: string }[]>` + select referenced.url + from migration_referenced_url referenced + left join "StorageMigrationTask" task + on task."sourceUrl" = referenced.url + and task."sourceStorageConfigId" = ${source.id} + where task."id" is null + order by referenced.url + limit 30 + `; + const unreferencedTasks = await transaction<{ count: string }[]>` + select count(*)::text as "count" + from "StorageMigrationTask" task + left join migration_referenced_url referenced on referenced.url = task."sourceUrl" + where task."status" in ('pending', 'failed', 'copying', 'copied') + and task."sourceStorageConfigId" = ${source.id} + and task."sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and task."sourcePath" like 'uploads/%' + and task."sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and referenced.url is null + `; + console.log("reference_coverage", { + ...coverage[0], + unreferencedActiveTasks: Number(unreferencedTasks[0]?.count || 0), + missingTaskSamples: missingSamples.map((item) => item.url), + }); + }); +} + +async function enqueueMissingReferences( + sql: ReturnType, + source: ReturnType +) { + const sourcePrefix = `${SOURCE_PREFIX}/uploads/`; + const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN; + const excludedTables = new Set([ + "StorageConfig", + "StorageMigrationTask", + "SystemSettings", + "SystemSettingHistory", + "SystemSettingSnapshot", + "__drizzle_migrations", + ]); + const columns = await sql< + { tableName: string; columnName: string; dataType: string }[] + >` + select + table_name as "tableName", + column_name as "columnName", + data_type as "dataType" + from information_schema.columns + where table_schema = 'public' + and data_type in ('text', 'character varying', 'json', 'jsonb') + order by table_name, ordinal_position + `; + const candidates = columns.filter((item) => !excludedTables.has(item.tableName)); + + await sql.begin(async (transaction) => { + await transaction.unsafe( + "create temporary table migration_referenced_url (url text primary key) on commit drop" + ); + for (const item of candidates) { + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + await transaction.unsafe( + ` + insert into migration_referenced_url (url) + select distinct (matches_array)[1] + from ${table} + cross join lateral regexp_matches(${column}::text, $1, 'gi') + as extracted(matches_array) + where ${column}::text like $2 + on conflict (url) do nothing + `, + [sourcePattern, `%${sourcePrefix}%`] + ); + } + const inserted = await transaction<{ id: string }[]>` + insert into "StorageMigrationTask" ( + "sourceUrl", + "sourcePath", + "sourceStorageConfigId", + "status", + "migrateAfterAt", + "cleanupAfterAt", + "updatedAt" + ) + select + referenced.url, + replace(referenced.url, ${`${SOURCE_PREFIX}/`}, ''), + ${source.id}, + 'pending', + now(), + now() + interval '30 days', + now() + from migration_referenced_url referenced + left join "StorageMigrationTask" task on task."sourceUrl" = referenced.url + where task."id" is null + on conflict ("sourceUrl") do nothing + returning "id" + `; + console.log("missing_references_enqueued", { inserted: inserted.length }); + }); +} + +async function replaceSampleReference( + sql: ReturnType, + source: ReturnType, + target: ReturnType +) { + const tasks = await sql< + { id: string; sourceUrl: string; targetUrl: string; sourcePath: string }[] + >` + select task."id", task."sourceUrl", task."targetUrl", task."sourcePath" + from "StorageMigrationTask" task + where task."status" = 'copied' + and task."sourceStorageConfigId" = ${source.id} + and task."sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and task."sourcePath" like 'uploads/%' + and task."sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and task."targetStorageConfigId" = ${target.id} + and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + and ( + exists ( + select 1 from "GeneratedImage" image + where image."originalUrl" = task."sourceUrl" + or image."thumbnailUrl" = task."sourceUrl" + ) + or exists ( + select 1 from "MediaAsset" asset + where asset."originalUrl" = task."sourceUrl" + or asset."normalizedUrl" = lower(task."sourceUrl") + or asset."previewUrl" = task."sourceUrl" + or asset."thumbnailUrl" = task."sourceUrl" + ) + ) + order by task."updatedAt" + limit 1 + `; + const task = tasks[0]; + if (!task) throw new Error("No copied and referenced task is available for sample replacement"); + + const targetHead = await retry(() => headPublicObject(task.targetUrl), 5); + if (!targetHead || targetHead.contentLength <= 0) { + throw new Error(`Sample target is not publicly readable: ${task.targetUrl}`); + } + + const excludedTables = new Set([ + "StorageConfig", + "StorageMigrationTask", + "SystemSettings", + "SystemSettingHistory", + "SystemSettingSnapshot", + "__drizzle_migrations", + ]); + const columns = await sql< + { tableName: string; columnName: string; dataType: string }[] + >` + select + table_name as "tableName", + column_name as "columnName", + data_type as "dataType" + from information_schema.columns + where table_schema = 'public' + and data_type in ('text', 'character varying', 'json', 'jsonb') + order by table_name, ordinal_position + `; + const candidates = columns.filter((item) => !excludedTables.has(item.tableName)); + + const result = await sql.begin(async (transaction) => { + const changedColumns: Array<{ + tableName: string; + columnName: string; + changedRows: number; + }> = []; + for (const item of candidates) { + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + const replacement = + item.dataType === "json" || item.dataType === "jsonb" + ? `replace(${column}::text, $1, $2)::${item.dataType}` + : `replace(${column}, $1, $2)`; + const rows = await transaction.unsafe<{ count: string }[]>( + ` + with updated as ( + update ${table} + set ${column} = ${replacement} + where ${column}::text like $3 + returning 1 + ) + select count(*)::text as count from updated + `, + [task.sourceUrl, task.targetUrl, `%${task.sourceUrl}%`] + ); + const changedRows = Number(rows[0]?.count || 0); + if (changedRows > 0) { + changedColumns.push({ + tableName: item.tableName, + columnName: item.columnName, + changedRows, + }); + } + } + await transaction` + update "StorageMigrationTask" + set "status" = 'migrated', + "migratedAt" = now(), + "cleanupAfterAt" = now() + interval '30 days', + "lastError" = null, + "updatedAt" = now() + where "id" = ${task.id} + and "status" = 'copied' + `; + return changedColumns; + }); + + const remaining = await sql<{ count: string }[]>` + select ( + select count(*) from "GeneratedImage" + where "originalUrl" = ${task.sourceUrl} or "thumbnailUrl" = ${task.sourceUrl} + ) + ( + select count(*) from "MediaAsset" + where "originalUrl" = ${task.sourceUrl} + or "previewUrl" = ${task.sourceUrl} + or "thumbnailUrl" = ${task.sourceUrl} + or "normalizedUrl" = ${task.sourceUrl.toLowerCase()} + ) as "count" + `; + console.log("sample_reference_replaced", { + taskId: task.id, + sourcePath: task.sourcePath, + targetBytes: targetHead.contentLength, + changedColumns: result, + directSourceReferencesRemaining: Number(remaining[0]?.count || 0), + }); +} + +async function replaceReadyReferences( + sql: ReturnType, + source: ReturnType, + target: ReturnType +) { + const sourcePrefix = `${SOURCE_PREFIX}/uploads/`; + const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN; + const targetReplacement = `${cleanUrl(target.publicUrl)}/\\1`; + const excludedTables = new Set([ + "StorageConfig", + "StorageMigrationTask", + "SystemSettings", + "SystemSettingHistory", + "SystemSettingSnapshot", + "__drizzle_migrations", + ]); + const columns = await sql< + { tableName: string; columnName: string; dataType: string }[] + >` + select + column_info.table_name as "tableName", + column_info.column_name as "columnName", + column_info.data_type as "dataType" + from information_schema.columns column_info + join information_schema.tables table_info + on table_info.table_schema = column_info.table_schema + and table_info.table_name = column_info.table_name + where column_info.table_schema = 'public' + and table_info.table_type = 'BASE TABLE' + and column_info.is_generated = 'NEVER' + and column_info.is_updatable = 'YES' + and column_info.data_type in ('text', 'character varying', 'json', 'jsonb') + order by column_info.table_name, column_info.ordinal_position + `; + const candidates = columns.filter((item) => !excludedTables.has(item.tableName)); + + const readiness = await sql.begin(async (transaction) => { + await transaction.unsafe( + "create temporary table migration_referenced_url (url text primary key) on commit drop" + ); + for (const item of candidates) { + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + await transaction.unsafe( + ` + insert into migration_referenced_url (url) + select distinct (matches_array)[1] + from ${table} + cross join lateral regexp_matches(${column}::text, $1, 'gi') + as extracted(matches_array) + where ${column}::text like $2 + on conflict (url) do nothing + `, + [sourcePattern, `%${sourcePrefix}%`] + ); + } + + const rows = await transaction< + { referenced: string; ready: string; unresolved: string }[] + >` + select + count(*)::text as "referenced", + count(*) filter ( + where task."status" in ('copied', 'migrated', 'deleted') + and task."targetStorageConfigId" = ${target.id} + and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + )::text as "ready", + count(*) filter ( + where task."id" is null + or task."status" not in ('copied', 'migrated', 'deleted') + or task."targetStorageConfigId" is distinct from ${target.id} + or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + )::text as "unresolved" + from migration_referenced_url referenced + left join "StorageMigrationTask" task + on task."sourceUrl" = referenced.url + and task."sourceStorageConfigId" = ${source.id} + `; + const unresolvedSamples = await transaction< + { url: string; status: string | null; lastError: string | null }[] + >` + select referenced.url, task."status", task."lastError" + from migration_referenced_url referenced + left join "StorageMigrationTask" task + on task."sourceUrl" = referenced.url + and task."sourceStorageConfigId" = ${source.id} + where task."id" is null + or task."status" not in ('copied', 'migrated', 'deleted') + or task."targetStorageConfigId" is distinct from ${target.id} + or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath" + order by referenced.url + limit 30 + `; + return { + referenced: Number(rows[0]?.referenced || 0), + ready: Number(rows[0]?.ready || 0), + unresolved: Number(rows[0]?.unresolved || 0), + unresolvedSamples, + }; + }); + + console.log("reference_replacement_readiness", readiness); + if (readiness.unresolved > 0) { + throw new Error( + `Refusing database URL replacement: ${readiness.unresolved} referenced objects are unresolved` + ); + } + + const changedColumns: Array<{ + tableName: string; + columnName: string; + changedRows: number; + }> = []; + for (const item of candidates) { + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + const replacement = + item.dataType === "json" || item.dataType === "jsonb" + ? `regexp_replace(${column}::text, $1, $2, 'gi')::${item.dataType}` + : `regexp_replace(${column}, $1, $2, 'gi')`; + const rows = await sql.begin(async (transaction) => { + await transaction.unsafe("set local lock_timeout = '3s'"); + await transaction.unsafe("set local statement_timeout = '15min'"); + return transaction.unsafe<{ count: string }[]>( + ` + with updated as ( + update ${table} + set ${column} = ${replacement} + where ${column}::text like $3 + returning 1 + ) + select count(*)::text as count from updated + `, + [ + PROJECT_IMAGE_SOURCE_PATH_CAPTURE_PATTERN, + targetReplacement, + `%${sourcePrefix}%`, + ] + ); + }); + const changedRows = Number(rows[0]?.count || 0); + if (changedRows > 0) { + const changed = { + tableName: item.tableName, + columnName: item.columnName, + changedRows, + }; + changedColumns.push(changed); + console.log("reference_column_replaced", changed); + } + } + + const remainingColumns: Array<{ + tableName: string; + columnName: string; + count: number; + }> = []; + for (const item of candidates) { + const table = quoteIdentifier(item.tableName); + const column = quoteIdentifier(item.columnName); + const rows = await sql.unsafe<{ count: string }[]>( + `select count(*)::text as count from ${table} where ${column}::text ~* $1`, + [PROJECT_IMAGE_SOURCE_URL_PATTERN] + ); + const count = Number(rows[0]?.count || 0); + if (count > 0) { + remainingColumns.push({ + tableName: item.tableName, + columnName: item.columnName, + count, + }); + } + } + if (remainingColumns.length > 0) { + console.log("reference_replacement_remaining", remainingColumns); + throw new Error( + `Database URL replacement is incomplete in ${remainingColumns.length} columns` + ); + } + + const migrated = await sql<{ id: string }[]>` + update "StorageMigrationTask" + set "status" = 'migrated', + "migratedAt" = coalesce("migratedAt", now()), + "cleanupAfterAt" = greatest( + coalesce("cleanupAfterAt", now()), + now() + interval '30 days' + ), + "lastError" = null, + "updatedAt" = now() + where "sourceStorageConfigId" = ${source.id} + and "targetStorageConfigId" = ${target.id} + and "status" = 'copied' + and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`} + and "sourcePath" like 'uploads/%' + and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN} + and "targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || "sourcePath" + returning "id" + `; + console.log("reference_replacement_finished", { + referencedUrls: readiness.referenced, + changedColumns, + changedRows: changedColumns.reduce((sum, item) => sum + item.changedRows, 0), + tasksMarkedMigrated: migrated.length, + sourceRetentionDays: 30, + }); +} + +async function main() { + if (!process.env.POSTGRES_URL) throw new Error("POSTGRES_URL is required"); + const concurrency = readIntArg("db-connections", 8, 2, 20); + const sql = postgres(process.env.POSTGRES_URL, { + max: concurrency, + idle_timeout: 30, + connect_timeout: 20, + prepare: false, + }); + try { + const { source, target } = await loadStorageConfigs(sql); + if (command === "audit") { + await audit(sql, source, target); + } else if (command === "probe") { + await probe(sql, source, target); + } else if (command === "copy") { + await copyTasks(sql, source, target); + } else if (command === "inventory-db") { + await inventoryDatabaseReferences(sql); + } else if (command === "audit-reference-coverage") { + await auditReferenceCoverage(sql, source, target); + } else if (command === "enqueue-missing-references") { + await enqueueMissingReferences(sql, source); + } else if (command === "replace-sample") { + await replaceSampleReference(sql, source, target); + } else if (command === "replace-ready-references") { + await replaceReadyReferences(sql, source, target); + } else { + throw new Error(`Unknown command: ${command}`); + } + } finally { + await sql.end({ timeout: 10 }); + } +} + +main().catch((error) => { + console.error("migration_failed", errorMessage(error)); + process.exitCode = 1; +}); diff --git a/scripts/storage/migrate-aliyun-to-qiniu.cjs b/scripts/storage/migrate-aliyun-to-qiniu.cjs new file mode 100644 index 000000000..629ede4e8 --- /dev/null +++ b/scripts/storage/migrate-aliyun-to-qiniu.cjs @@ -0,0 +1,220 @@ +/* Resumable, verified migration. No source objects are deleted. */ +const fs = require('node:fs'); +const path = require('node:path'); +const crypto = require('node:crypto'); +const { Readable, Transform } = require('node:stream'); +const SOURCE_HOST = 'demo-sparkai.oss-cn-shanghai.aliyuncs.com'; +const URL_PATTERN = /https?:\/\/(?:demo-sparkai\.oss-cn-shanghai\.aliyuncs\.com\/|s3\.jiansun\.vip\/gongxiangpan\/)[^\s"'<>`\\)\]}|]+/gi; +function resourceFor(url) { + const u = new URL(url); + const isPrimary=u.hostname===SOURCE_HOST,isPsdBackup=u.hostname==='s3.jiansun.vip'&&u.pathname.startsWith('/gongxiangpan/'); + if ((!isPrimary&&!isPsdBackup) || u.pathname === '/') return null; + for (const name of ['Signature', 'OSSAccessKeyId', 'Expires', 'security-token']) u.searchParams.delete(name); + u.protocol = 'https:'; u.hash = ''; u.searchParams.sort(); + const identity = u.toString(), sourceKey = decodeURIComponent(u.pathname.slice(isPsdBackup?'/gongxiangpan/'.length:1)); + if (!sourceKey || sourceKey.split('/').some(p => p === '..')) throw Error('Invalid object key'); + const id = crypto.createHash('sha256').update(identity).digest('hex'); + const namespace=isPrimary?'legacy-oss/demo-sparkai/':'legacy-s3/gongxiangpan/'; + const targetKey = u.search ? namespace+'variants/' + id + path.extname(sourceKey) : namespace + sourceKey; + return { id, identity, sourceHost:u.hostname, sourceKey, targetKey, variant: Boolean(u.search) }; +} +function urlsIn(value) { + const found = new Map(); + function walk(v) { + if (typeof v === 'string') for (const url of v.match(URL_PATTERN) || []) { const r = resourceFor(url); if (r) found.set(r.id, r); } + else if (Array.isArray(v)) v.forEach(walk); + else if (v && typeof v === 'object') Object.entries(v).forEach(([k,val])=>{walk(k);walk(val);}); + } + walk(value); return [...found.values()]; +} +function replaceVerified(value, verified) { + if (typeof value === 'string') return value.replace(URL_PATTERN, url => { const r = resourceFor(url); return r && verified.get(r.id)?.targetUrl || url; }); + if (Array.isArray(value)) return value.map(v => replaceVerified(v, verified)); + if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([k, v]) => [replaceVerified(k, verified), replaceVerified(v, verified)])); + return value; +} +function quote(s) { return '"' + s.replaceAll('"', '""') + '"'; } +async function rewriteBatch(sql, column, pk, pkType, rows, verified, backup) { + const isJson=['json','jsonb'].includes(column.dataType); + const changes=[],originals=[]; + for(const row of rows){ + const next=replaceVerified(row.value,verified); + if(JSON.stringify(next)===JSON.stringify(row.value))continue; + originals.push({table:column.tableName,column:column.columnName,pkColumn:pk,pk:row.pk,value:row.value}); + changes.push({pk:String(row.pk),oldvalue:isJson?JSON.stringify(row.value):row.value,newvalue:isJson?JSON.stringify(next):next}); + } + if(!changes.length)return 0; + backup(originals); + const cast=isJson?'::'+column.dataType:''; + const updated=await sql.unsafe('update '+quote(column.tableName)+' as t set '+quote(column.columnName)+'=v.newvalue'+cast+' from jsonb_to_recordset($1::jsonb) as v(pk text,oldvalue text,newvalue text) where t.'+quote(pk)+'=v.pk::'+pkType+' and t.'+quote(column.columnName)+(isJson?'::jsonb=v.oldvalue::jsonb':'=v.oldvalue')+' returning t.'+quote(pk),[sql.json(changes)]); + return updated.length; +} +module.exports = { resourceFor, urlsIn, replaceVerified, rewriteBatch }; +if (require.main === module) main().catch(e => { console.error('migration_failed', e.name, String(e.message).replace(/https?:\/\/\S+/g, '[URL]').slice(0,240)); process.exitCode = 1; }); +async function main() { + const root = process.env.MIGRATION_APP_ROOT || process.cwd(); + require(path.join(root, 'node_modules/dotenv')).config({ path: path.join(root, '.env.local') }); + const postgres = require(path.join(root, 'node_modules/postgres')); + const { S3Client, PutObjectCommand, GetObjectCommand, HeadObjectCommand } = require(path.join(root, 'node_modules/@aws-sdk/client-s3')); + const sql = postgres(process.env.POSTGRES_URL, { max: 2, prepare: false, connection: { statement_timeout: 600000 } }); + const dir = process.env.MIGRATION_RUN_DIR || '/var/lib/ai-chatbot/aliyun-qiniu-20261014'; + fs.mkdirSync(dir, { recursive: true, mode: 0o700 }); + const journal = path.join(dir, 'verified.jsonl'), verified = new Map(); + if (fs.existsSync(journal)) for (const line of fs.readFileSync(journal,'utf8').split('\n').filter(Boolean)) { const r = JSON.parse(line); verified.set(r.id,r); } + let sourceClient, targetClient; + try { + const configs = await sql`select * from "StorageConfig"`; + const source = configs.find(c => c.bucket === 'demo-sparkai' && String(c.endpoint).includes('aliyuncs.com')); + const target = configs.find(c => c.type === 'qiniu-kodo' && c.isEnabled && c.isDefault); + if (!source || !target || !target.publicUrl) throw Error('Expected source and default Qiniu configuration unavailable'); + function client(c) { return new S3Client({ endpoint: c.endpoint, region: c.region || 'auto', forcePathStyle: true, credentials: { accessKeyId:c.accessKeyId,secretAccessKey:c.secretAccessKey }, requestChecksumCalculation:'WHEN_REQUIRED',responseChecksumValidation:'WHEN_REQUIRED', maxAttempts: 2 }); } + sourceClient = client(source); targetClient = client(target); + const publicUrl = key => target.publicUrl.replace(/\/$/,'') + '/' + key.split('/').map(encodeURIComponent).join('/'); + const command = process.argv[2] || 'audit'; + const settings = command==='migrate-psd-extra'?await sql`select id,key,value from "SystemSettings" where key in ('agent_psd_templates','agent_psd_fonts')`:await sql`select id,key,value from "SystemSettings" where value::text like ${'%'+SOURCE_HOST+'/%'}`; + const settingsResources = urlsIn(settings.map(s => s.value)); + console.log('settings_inventory', JSON.stringify({ rows: settings.length, keys: settings.map(s => s.key), uniqueObjects:settingsResources.length, alreadyVerified:verified.size })); + if (command === 'audit') return; + const resources = new Map(settingsResources.map(r => [r.id,r])); + const scope = command.endsWith('-all') ? 'all' : command==='migrate-psd-extra'?'psd-extra':command==='migrate-recovery'?'recovery':command==='migrate-literals'?'literals':'settings'; + if(command==='migrate-recovery'){ + resources.clear(); + for(const candidate of JSON.parse(fs.readFileSync(process.env.MIGRATION_RESOURCE_FILE,'utf8'))){ + const canonical=resourceFor(candidate.identity); + if(!canonical||canonical.id!==candidate.id)throw Error('Invalid recovery identity'); + const backup=new URL(candidate.fetchUrl); + if(!configs.some(c=>c.publicUrl&&backup.origin===new URL(c.publicUrl).origin&&backup.pathname.startsWith(new URL(c.publicUrl).pathname.replace(/\/$/,'')+'/')))throw Error('Recovery source is not a configured storage origin'); + resources.set(canonical.id,{...canonical,fetchUrl:candidate.fetchUrl}); + } + } + let columns = []; + if (scope === 'all') { + columns = await sql`select c.table_name as "tableName",c.column_name as "columnName",c.data_type as "dataType" from information_schema.columns c join information_schema.tables t on c.table_schema=t.table_schema and c.table_name=t.table_name where c.table_schema='public' and t.table_type='BASE TABLE' and c.is_generated='NEVER' and c.is_updatable='YES' and c.data_type in ('text','character varying','json','jsonb') order by c.table_name,c.ordinal_position`; + const excluded = new Set(['StorageConfig','StorageMigrationTask','ImageArchiveJob','__drizzle_migrations','SystemSettings','SystemSettingHistory','SystemSettingSnapshot']); + const tables = new Map(); + const inventoryJournal=path.join(dir,'inventory.jsonl'), doneFile=path.join(dir,'inventory-completed-tables.json'); + const completedTables=new Set(fs.existsSync(doneFile)?JSON.parse(fs.readFileSync(doneFile,'utf8')):[]); + if(fs.existsSync(inventoryJournal))for(const line of fs.readFileSync(inventoryJournal,'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line);for(const canonical of urlsIn(r.identity))resources.set(canonical.id,canonical);} + for (const c of columns) if (!excluded.has(c.tableName)) { if (!tables.has(c.tableName)) tables.set(c.tableName,[]); tables.get(c.tableName).push(c); } + for (const [table, cols] of tables) { + if(completedTables.has(table)&&process.env.MIGRATION_RESCAN!=='1')continue; + console.log('scan_start',table); + const where = cols.map(c => quote(c.columnName)+'::text like $1').join(' or '); + let matched=0; + for await (const rows of sql.unsafe('select '+cols.map(c=>quote(c.columnName)).join(',')+' from '+quote(table)+' where '+where,['%'+SOURCE_HOST+'/%']).cursor(50)){const added=[];for (const row of rows) { matched++; for (const r of urlsIn(row)) if(!resources.has(r.id)){resources.set(r.id,r);added.push(r);} }if(added.length)fs.appendFileSync(inventoryJournal,added.map(r=>JSON.stringify(r)).join('\n')+'\n',{mode:0o600});} + completedTables.add(table);fs.writeFileSync(doneFile,JSON.stringify([...completedTables]),{mode:0o600}); + if (matched) console.log('database_inventory',JSON.stringify({table,rows:matched,uniqueObjects:resources.size})); + } + } + if(scope==='all'||scope==='literals'){ + if(scope==='literals')resources.clear(); + for (const folder of ['app','components','lib']) { + function walk(d) { for (const e of fs.readdirSync(d,{withFileTypes:true})) { const f=path.join(d,e.name); if(e.isDirectory())walk(f); else if(/\.(tsx?|jsx?)$/.test(f)) for(const r of urlsIn(fs.readFileSync(f,'utf8')))resources.set(r.id,r); } } + walk(path.join(root,folder)); + } + } + fs.writeFileSync(path.join(dir,'plan-'+scope+'.json'),JSON.stringify([...resources.values()]),{mode:0o600}); + if (command === 'inventory-all') return; + if (!['migrate-settings','migrate-all','migrate-psd-extra','migrate-recovery','migrate-literals','rewrite-all'].includes(command)) throw Error('Unknown command'); + if (scope === 'settings' && !fs.existsSync(path.join(dir,'settings-before.json'))) fs.writeFileSync(path.join(dir,'settings-before.json'),JSON.stringify(settings),{mode:0o600,flag:'wx'}); + const sourceMetadata=new Map(); const metadataFile=path.join(dir,'source-metadata.jsonl'); + if(fs.existsSync(metadataFile))for(const line of fs.readFileSync(metadataFile,'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line);sourceMetadata.set(r.key,r);} + const base64url = value => Buffer.from(value).toString('base64').replace(/\+/g,'-').replace(/\//g,'_'); + async function cloudCopy(r) { + const fetchUrl=r.fetchUrl||r.identity; + const cached=r.fetchUrl||r.variant||new URL(r.identity).hostname!==SOURCE_HOST?null:sourceMetadata.get(r.sourceKey); + let size=cached?.size,md5=cached?.md5; + if(!r.variant&&!cached){const head=await fetch(fetchUrl,{method:'HEAD',headers:{'Accept-Encoding':'identity'},signal:AbortSignal.timeout(45000)});if(!head.ok)throw Error('Source HTTP '+head.status);size=Number(head.headers.get('content-length'));md5=String(head.headers.get('etag')||'').replaceAll('"','').toLowerCase();} + if(r.variant || !size || !/^[a-f0-9]{32}$/.test(md5||'')) { + const download=await fetch(fetchUrl,{headers:{'Accept-Encoding':'identity'},signal:AbortSignal.timeout(15*60*1000)}); + if(!download.ok)throw Error('Source HTTP '+download.status); + const hash=crypto.createHash('md5');size=0;for await(const chunk of Readable.fromWeb(download.body)){hash.update(chunk);size+=chunk.length;}md5=hash.digest('hex'); + } + if(!size)throw Error('Empty source object'); + const requestPath='/fetch/'+base64url(fetchUrl)+'/to/'+base64url(target.bucket+':'+r.targetKey); + const signature=base64url(crypto.createHmac('sha1',target.secretAccessKey).update(requestPath+'\n').digest()); + const fetchResult=await fetch('https://iovip.qbox.me'+requestPath,{method:'POST',headers:{Authorization:'QBox '+target.accessKeyId+':'+signature,'Content-Type':'application/x-www-form-urlencoded'},signal:AbortSignal.timeout(15*60*1000)}); + if(!fetchResult.ok)throw Error('Qiniu cloud fetch HTTP '+fetchResult.status); + const info=await fetchResult.json();if(Number(info.fsize)!==size)throw Error('Cloud target size mismatch'); + const targetUrl=publicUrl(r.targetKey); + const hashResponse=await fetch(targetUrl+'?qhash/md5',{signal:AbortSignal.timeout(45000)}); + if(!hashResponse.ok)throw Error('CDN hash verification HTTP '+hashResponse.status); + const targetDigest=await hashResponse.json(); + if(targetDigest.hash!==md5 || Number(targetDigest.fsize)!==size)throw Error('Cloud target checksum mismatch'); + const record={...r,targetUrl,size,md5,transport:'qiniu-cloud-fetch',verifiedAt:new Date().toISOString()};fs.appendFileSync(journal,JSON.stringify(record)+'\n',{mode:0o600});verified.set(r.id,record); + } + async function copy(r) { + if (verified.has(r.id)) return; + let last; + for (let attempt=1;attempt<=3;attempt++) { + let input; + try { + if(process.env.MIGRATION_TRANSPORT!=='stream'){await cloudCopy(r);return;} + const timeout=AbortSignal.timeout(15*60*1000); + const response=await fetch(r.identity,{signal:timeout,headers:{'Accept-Encoding':'identity'}}); + let size,type,disposition; + if (response.ok) { size=Number(response.headers.get('content-length'));type=response.headers.get('content-type');disposition=response.headers.get('content-disposition'); input=Readable.fromWeb(response.body); } + else { + await response.body?.cancel(); + if(r.variant)throw Error('Source variant HTTP '+response.status); + const get=await sourceClient.send(new GetObjectCommand({Bucket:source.bucket,Key:r.sourceKey}),{abortSignal:timeout});input=get.Body;size=get.ContentLength;type=get.ContentType;disposition=get.ContentDisposition; + } + if (!size || size>5*1024**3)throw Error('Unsupported source size '+size); + const hash=crypto.createHash('sha256');let readBytes=0; + const hashStream=new Transform({transform(chunk,enc,cb){hash.update(chunk);readBytes+=chunk.length;cb(null,chunk)}}); + input.on('error',e=>hashStream.destroy(e)); + await targetClient.send(new PutObjectCommand({Bucket:target.bucket,Key:r.targetKey,Body:input.pipe(hashStream),ContentLength:size,ContentType:type||'application/octet-stream',ContentDisposition:disposition||undefined,CacheControl:'public, max-age=31536000, immutable',Metadata:{'migration-source-id':r.id}}),{abortSignal:timeout}); + if(readBytes!==size)throw Error('Source size mismatch'); const sha256=hash.digest('hex'); + const get=await targetClient.send(new GetObjectCommand({Bucket:target.bucket,Key:r.targetKey}),{abortSignal:timeout});const targetHash=crypto.createHash('sha256');let storedBytes=0;for await(const chunk of get.Body){targetHash.update(chunk);storedBytes+=chunk.length;} + if(storedBytes!==size||targetHash.digest('hex')!==sha256)throw Error('Target checksum mismatch'); + const targetUrl=publicUrl(r.targetKey);const head=await fetch(targetUrl,{method:'HEAD',signal:AbortSignal.timeout(30000)}); + if(!head.ok||Number(head.headers.get('content-length'))!==size)throw Error('CDN verification HTTP '+head.status); + const record={...r,targetUrl,size,sha256,verifiedAt:new Date().toISOString()};fs.appendFileSync(journal,JSON.stringify(record)+'\n',{mode:0o600});verified.set(r.id,record);return; + } catch(e) { last=e;input?.destroy();if(attempt<3)await new Promise(resolve=>setTimeout(resolve,attempt*1000)); } + } + fs.appendFileSync(path.join(dir,'failures-'+scope+'.jsonl'),JSON.stringify({id:r.id,sourceKey:r.sourceKey,name:last.name,message:String(last.message).replace(/https?:\/\/\S+/g,'[URL]').slice(0,240)})+'\n',{mode:0o600}); + } + const psdIds=new Set(urlsIn(settings.find(s=>s.key==='agent_psd_templates')?.value).map(r=>r.id)); + const list=[...resources.values()].sort((a,b)=>(psdIds.has(a.id)?0:2)-(psdIds.has(b.id)?0:2)||Number(/\.(psd|psb)$/i.test(a.sourceKey))-Number(/\.(psd|psb)$/i.test(b.sourceKey))); + const pending=list.filter(r=>!verified.has(r.id)); + const concurrency=Math.max(1,Math.min(96,Number(process.env.MIGRATION_CONCURRENCY||64))); + let cursor=0,done=0;const start=Date.now(); + function progress(){const p={scope,total:list.length,processed:done,verified:verified.size,remaining:list.filter(r=>!verified.has(r.id)).length,elapsedSeconds:Math.round((Date.now()-start)/1000),verifiedBytes:[...verified.values()].reduce((n,r)=>n+r.size,0)};fs.writeFileSync(path.join(dir,'progress-'+scope+'.json'),JSON.stringify(p),{mode:0o600});console.log('copy_progress',JSON.stringify(p));} + const interval=setInterval(progress,30000); + try { if(command!=='rewrite-all')await Promise.all(Array.from({length:concurrency},async()=>{while(cursor!verified.has(r.id)); + // Only verified addresses are changed; unresolved references keep their original source. + let changedSettings=0; + for(const setting of settings){ + for(let attempt=0;attempt<4;attempt++){ + const [current]=await sql`select value from "SystemSettings" where id=${setting.id}`;const next=replaceVerified(current.value,verified);if(JSON.stringify(next)===JSON.stringify(current.value))break; + const updated=await sql`update "SystemSettings" set value=${sql.json(next)},"updatedAt"=now() where id=${setting.id} and value=${sql.json(current.value)} returning id`; + if(updated.length){changedSettings++;break;}if(attempt===3)throw Error('Concurrent setting update prevented replacement'); + } + } + if(scope==='all'){ + const backupFile=path.join(dir,'database-before.jsonl.gz'); + for(const c of columns){ + // Administrative audit snapshots preserve their original configuration values. + // User canvas/chat history remains in scope; obsolete configuration backups do not. + if(['StorageConfig','StorageMigrationTask','ImageArchiveJob','__drizzle_migrations','SystemSettings','SystemSettingHistory','SystemSettingSnapshot'].includes(c.tableName))continue; + const pks=await sql`select a.attname,a.atttypid::regtype::text as "pkType" from pg_index i join pg_attribute a on a.attrelid=i.indrelid and a.attnum=any(i.indkey) where i.indrelid=${'"'+c.tableName+'"'}::regclass and i.indisprimary`; + if(pks.length!==1){continue;} const pk=pks[0].attname; + let count=0; + const aliasGuard=c.tableName==='ImageArchiveTask'&&c.columnName==='sourceUrl'?" and coalesce(metadata->>'legacyOssMigrationRun','') <> '20261014'":''; + for await(const rows of sql.unsafe('select '+quote(pk)+' as pk,'+quote(c.columnName)+' as value from '+quote(c.tableName)+' where '+quote(c.columnName)+'::text like $1'+aliasGuard,['%'+SOURCE_HOST+'/%']).cursor(50)){ + // Flush before updating to retain the original data even on interruption. + count+=await rewriteBatch(sql,c,pk,pks[0].pkType,rows,verified,backups=>fs.appendFileSync(backupFile,require('node:zlib').gzipSync(backups.map(r=>JSON.stringify(r)).join('\n')+'\n',{level:6}),{mode:0o600})); + } + if(count)console.log('references_replaced',JSON.stringify({table:c.tableName,column:c.columnName,rows:count})); + } + } + const [migration]=await sql`select id,value from "SystemSettings" where key='oss_image_migration_config'`; + if(migration){const next={...migration.value,enabled:false,preferAliyunForUploads:false,targetStorageConfigId:target.id};await sql`update "SystemSettings" set value=${sql.json(next)},"updatedAt"=now() where id=${migration.id}`;} + await sql`update "StorageConfig" set "isEnabled"=false,"isDefault"=false where id=${source.id}`; + const result={scope,objects:list.length,verified:list.length-missing.length,missing:missing.length,changedSettings,defaultStorage:'qiniu-kodo',sourceRetained:true,completedAt:new Date().toISOString()};fs.writeFileSync(path.join(dir,'result-'+scope+'.json'),JSON.stringify(result,null,2));console.log('migration_result',JSON.stringify(result)); + if(missing.length)process.exitCode=2; + } finally {sourceClient?.destroy();targetClient?.destroy();await sql.end();} +} diff --git a/scripts/storage/register-aliyun-qiniu-aliases.cjs b/scripts/storage/register-aliyun-qiniu-aliases.cjs new file mode 100644 index 000000000..52f1aa3f8 --- /dev/null +++ b/scripts/storage/register-aliyun-qiniu-aliases.cjs @@ -0,0 +1,6 @@ +/* Publish verified legacy aliases through the application's existing completed-archive resolver. + * No source objects are deleted, and no pending archive jobs are created or overwritten. */ +const fs=require('fs'),path=require('path'),crypto=require('crypto');const root=process.env.MIGRATION_APP_ROOT||process.cwd(),dir=process.env.MIGRATION_RUN_DIR||'/var/lib/ai-chatbot/aliyun-qiniu-20261014';const {resourceFor}=require('./migrate-aliyun-to-qiniu.cjs');require(path.join(root,'node_modules/dotenv')).config({path:path.join(root,'.env.local')});const sql=require(path.join(root,'node_modules/postgres'))(process.env.POSTGRES_URL,{max:1,prepare:false}); +(async()=>{try{const [target]=await sql`select id,bucket,"publicUrl" from "StorageConfig" where type='qiniu-kodo' and "isDefault" and "isEnabled" limit 1`;if(!target?.publicUrl)throw Error('No default Qiniu target');const map=new Map();for(const line of fs.readFileSync(path.join(dir,'verified.jsonl'),'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line),canonical=resourceFor(r.identity),expectedUrl=target.publicUrl.replace(/\/$/,'')+'/'+r.targetKey.split('/').map(encodeURIComponent).join('/');if(!canonical||canonical.id!==r.id||r.targetUrl!==expectedUrl||r.size<=0||(!r.md5&&!r.sha256))throw Error('Unverified storage alias');map.set(r.identity,r);}let inserted=0,processed=0,batch=[]; +async function flush(){if(!batch.length)return;const added=await sql.unsafe('insert into "ImageArchiveTask" (id,"sourceUrl","targetUrl","targetStorageConfigId",status,"attemptCount","archiveAfterAt","replaceAfterAt","archivedAt","replacedAt",metadata,"createdAt","updatedAt") select gen_random_uuid(),r.source,r.target,$2::uuid,\'replaced\',0,now(),now(),now(),now(),r.metadata,now(),now() from jsonb_to_recordset($1::jsonb) as r(source text,target text,metadata jsonb) on conflict ("sourceUrl") do nothing returning id',[sql.json(batch),target.id]);inserted+=added.length;processed+=batch.length;batch=[];if(processed%25000===0)console.log(JSON.stringify({processed,total:map.size,inserted}));} +for(const r of map.values()){batch.push({source:r.identity,target:r.targetUrl,metadata:{legacyOssMigrationRun:'20261014',migrationResourceId:r.id,migrationVerifiedAt:r.verifiedAt,verifiedBytes:r.size,verifiedChecksum:r.md5||r.sha256}});if(batch.length>=500)await flush();}await flush();const result={aliasResources:map.size,inserted,existingPreserved:map.size-inserted,pendingJobsCreated:0,completedAt:new Date().toISOString()};fs.writeFileSync(path.join(dir,'alias-registration.json'),JSON.stringify(result,null,2),{mode:0o600});console.log('alias_registration',JSON.stringify(result));}finally{await sql.end()}})().catch(e=>{console.error('alias_registration_failed',e.name,String(e.message).replace(/https?:\/\/\S+/g,'[URL]').slice(0,180));process.exitCode=1}); diff --git a/scripts/zcode-agent/native-fixture.mjs b/scripts/zcode-agent/native-fixture.mjs index e4e71f449..699dc3eff 100644 --- a/scripts/zcode-agent/native-fixture.mjs +++ b/scripts/zcode-agent/native-fixture.mjs @@ -31,6 +31,8 @@ try { onEvent: async event => { events.push(event.type); }, onSandbox: async sandbox => console.log('SANDBOX', JSON.stringify(sandbox)), model: async body => { + assert.notEqual(body.thinking?.type, 'enabled'); + assert(!body.reasoning_effort || ['none', 'disabled'].includes(body.reasoning_effort)); requests++; if (requests > 12) return Response.json({ error: { message: 'fixture budget exceeded' } }, { status: 429 }); const names = (body.tools || []).map(tool => tool.function?.name || tool.name || ''); console.log('MODEL', requests, 'tools', names.filter(name => /readCanvas|renameNode/.test(name))); diff --git a/scripts/zcode-agent/task-fixture.ts b/scripts/zcode-agent/task-fixture.ts index 565eac094..4d784657a 100644 --- a/scripts/zcode-agent/task-fixture.ts +++ b/scripts/zcode-agent/task-fixture.ts @@ -5,6 +5,7 @@ import { runZcodeCanvasTask } from '../../lib/ai/zcode-agent/task'; import { AgentToolRejectedError, newAgentRunSnapshot, type AgentRunSnapshot } from '../../lib/ai/canvas-agent-runtime/protocol'; import type { runNativeZcode } from '../../services/zcode-agent/runner.mjs'; import { AgentImageGenerationError } from '../../lib/ai/runtime/image-generation-error'; +import { canvasAgentReplyText } from '../../lib/ai/zcode-agent/assistant-output'; type NativeOptions = Parameters[0]; function fixture(objective = '修改图片') { const snapshot = newAgentRunSnapshot(objective, []); let writes = 0; const saved: string[] = []; @@ -24,6 +25,31 @@ function fixture(objective = '修改图片') { const result = { type: 'result', sessionId: 'sess_fixture', response: '已完成' }; async function main() { let checks = 0; +{ + const h=fixture('画布里有什么'); const emitted:string[]=[]; + const textPorts={...h.ports,onAssistantText:(delta:string)=>{emitted.push(delta);}}; + textPorts.native=async options=>{ + await options.onEvent?.({type:'turn.started',sessionId:'public-session'}); + await options.onEvent?.({type:'model.streaming',sessionId:'public-session',payload:{kind:'reasoning_delta',delta:'PRIVATE_THOUGHT'}}); + await options.onEvent?.({type:'model.streaming',sessionId:'child-session',payload:{kind:'text_delta',delta:'PRIVATE_CHILD'}}); + const event={type:'model.streaming',eventId:'text-1',sessionId:'public-session',payload:{kind:'text_delta',assistantMessageId:'a',delta:'画布里有一只小鸡。'}}; + await options.onEvent?.(event);await options.onEvent?.(event); + assert.equal(emitted.join(''),'画布里有一只小鸡。','ordinary reply must arrive before native completion'); + return {...result,response:'画布里有一只小鸡。'}; + }; + assert.equal(await runZcodeCanvasTask(textPorts),'completed'); + assert.equal(canvasAgentReplyText(h.snapshot),'画布里有一只小鸡。'); + assert.equal(emitted.length,1);checks++; +} +{ + const h=fixture('分析图片');h.ports.native=async options=>{ + await options.onEvent?.({type:'turn.started',sessionId:'session'}); + await options.onEvent?.({type:'model.streaming',sessionId:'session',payload:{kind:'text_delta',assistantMessageId:'a',delta:'已读取图片，正在核对。'}}); + throw new Error('连接中断'); + }; + assert.equal(await runZcodeCanvasTask(h.ports),'needs_review'); + assert.match(canvasAgentReplyText(h.snapshot),/^已读取图片，正在核对。\n\n连接中断$/);checks++; +} for (const passed of [true,false]) { const h=fixture();let paid=0,reviews=0; h.ports.tools.edit.execute=async()=>{paid++;return {updatedNodeIds:['a'],createdNodeIds:['real-result']};}; diff --git a/services/zcode-agent/runner.d.mts b/services/zcode-agent/runner.d.mts index f95dc1c6a..db842ba2b 100644 --- a/services/zcode-agent/runner.d.mts +++ b/services/zcode-agent/runner.d.mts @@ -1,6 +1,6 @@ export const UPSTREAM_COMMIT: string; export type NativeTool = { name: string; description: string; inputSchema: object; readOnly: boolean }; -export type NativeEvent = { type: string; sessionId?: string; traceId?: string; payload?: Record }; +export type NativeEvent = { type: string; eventId?: string; sessionId?: string; traceId?: string; payload?: Record }; export type NativeResult = NativeEvent & { sessionId: string; response: string; usage?: Record; projection?: Record }; export function verifyRuntime(runtimeDir: string): Promise<{ commit: string; sha256: string }>; export function runNativeZcode(options: { diff --git a/services/zcode-agent/runner.mjs b/services/zcode-agent/runner.mjs index 43e2bb6a8..7ab86a930 100644 --- a/services/zcode-agent/runner.mjs +++ b/services/zcode-agent/runner.mjs @@ -104,8 +104,9 @@ export async function runNativeZcode(options) { providerConfigRules: { providerRules: [{ providerId: 'canvas', providerName: 'Canvas Task Gateway', config: { group: 'standard-personal', access: { type: 'api-key', apiKey: token }, api: { type: options.apiType || 'openai-chat-completions', baseUrl }, personalModelIds: [options.modelId], - } }] }, modelConfigRules: { providerModelRules: [], manualProviderModelRules: [] }, - defaultModelSelection: { providerId: 'canvas', modelId: options.modelId }, + } }] }, modelConfigRules: { providerModelRules: [{ providerId: 'canvas', modelId: options.modelId, + config: { optionSpecs: { reasoningLevel: { values: ['disabled'], map: '{}' } } } }], manualProviderModelRules: [] }, + defaultModelSelection: { providerId: 'canvas', modelId: options.modelId, options: { reasoningLevel: 'disabled' } }, } }; await writeFile(join(home, '.zcode/v2/provider_config.json'), JSON.stringify(personal), { mode: 0o600 }); const bridge = linux ? '/runtime/bridge.mjs' : join(options.runtimeDir, 'bridge.mjs'); diff --git a/tests/integration/agent-model-gateway.test.ts b/tests/integration/agent-model-gateway.test.ts index fc554f10c..9324e3bfa 100644 --- a/tests/integration/agent-model-gateway.test.ts +++ b/tests/integration/agent-model-gateway.test.ts @@ -50,6 +50,21 @@ function fixture() { return { paused, calls, circuit }; } +test("Agent fallback strips native deep thinking and disables Qwen thinking", async () => { + const original=global.fetch;const requests:any[]=[]; + global.fetch=async (_url,init)=>{requests.push(JSON.parse(String(init?.body)));return Response.json({choices:[{finish_reason:'stop',message:{role:'assistant',content:'普通回复'}}]});}; + try { + for (const id of ['qwen3-vl-plus','generic-model']) { + const model=entry(id);const gateway=await createZcodeModelGateway(model,undefined,[model],fixture().circuit); + const response=await gateway.forward({...body,stream:false,thinking:{type:'enabled'},enable_thinking:true,reasoning_effort:'high',reasoning:{effort:'high'}},'/v1/chat/completions',new AbortController().signal); + assert.equal(response.status,200); + } + assert.equal(requests[0].enable_thinking,false); + for (const request of requests) {assert.equal(request.thinking,undefined);assert.equal(request.reasoning,undefined);assert.equal(request.reasoning_effort,undefined);} + assert.equal(requests[1].enable_thinking,undefined); + } finally {global.fetch=original;} +}); + test("a broken primary and a Responses-only backup work through one stable native protocol", async () => { const s = fixture(); const primary = entry("primary"); const backup = entry("backup", true); const original = global.fetch; diff --git a/tsconfig.json b/tsconfig.json index 795ac5463..c4b42eb9c 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,7 +1,11 @@ { "compilerOptions": { "target": "ESNext", - "lib": ["dom", "dom.iterable", "esnext"], + "lib": [ + "dom", + "dom.iterable", + "esnext" + ], "allowJs": true, "skipLibCheck": true, "strict": true, @@ -19,7 +23,9 @@ } ], "paths": { - "@/*": ["./*"] + "@/*": [ + "./*" + ] }, "strictNullChecks": true }, @@ -39,7 +45,15 @@ ".next-build-hicad-20260728-140320/types/**/*.ts", ".next-build-hicad-20260728-140320/dev/types/**/*.ts", ".next-candidate-home-banner-20260804-bg/types/**/*.ts", - ".next-candidate-home-banner-20260804-bg/dev/types/**/*.ts" + ".next-candidate-home-banner-20260804-bg/dev/types/**/*.ts", + ".next-extend-vip98k-20261012-1791792190533/types/**/*.ts", + ".next-extend-vip98k-20261012-1791792190533/dev/types/**/*.ts", + ".next-psd-dimensions-e6dbb612-20261015-1792035218252/types/**/*.ts", + ".next-psd-dimensions-e6dbb612-20261015-1792035218252/dev/types/**/*.ts", ], - "exclude": ["node_modules", "services/pascal-editor", "backups"] + "exclude": [ + "node_modules", + "services/pascal-editor", + "backups" + ] }