242 lines
20 KiB
TypeScript
242 lines
20 KiB
TypeScript
import "server-only";
|
||
import { asSchema, type ToolSet } from "ai";
|
||
import { join } from "node:path";
|
||
import type { NativeEvent, NativeResult } from "@/services/zcode-agent/runner.mjs";
|
||
import { runNativeZcode } from "./native-runtime";
|
||
import type { CanvasAgentTool } from "@/lib/ai/canvas-agent-runtime/loop";
|
||
import { AgentToolRejectedError, publicAgentError, type AgentRunSnapshot, type CanvasAgentRunStatus } from "@/lib/ai/canvas-agent-runtime/protocol";
|
||
import { isCompletedSimpleCutout } from "@/lib/ai/canvas-agent-runtime/cutout-completion";
|
||
import { isAncillaryDownload } from "@/lib/ai/canvas-agent-runtime/results";
|
||
|
||
type Ports = {
|
||
native?: typeof runNativeZcode;
|
||
runId: string;
|
||
snapshot: AgentRunSnapshot;
|
||
definitions: ToolSet;
|
||
tools: Record<string, CanvasAgentTool>;
|
||
instructions: string;
|
||
context: unknown;
|
||
gateway: { modelId: string; apiType: string; forward: (body: Record<string, unknown>, path: string, signal: AbortSignal) => Promise<Response> };
|
||
signal: AbortSignal;
|
||
checkpoint: (snapshot: AgentRunSnapshot, status: CanvasAgentRunStatus) => Promise<void>;
|
||
consumeInputs: (snapshot: AgentRunSnapshot) => Promise<{ cancelled: boolean }>;
|
||
isCancelled: () => Promise<boolean>;
|
||
verify: (snapshot: AgentRunSnapshot) => Promise<NonNullable<AgentRunSnapshot["verification"]>>;
|
||
};
|
||
|
||
export async function runZcodeCanvasTask(ports: Ports): Promise<CanvasAgentRunStatus> {
|
||
const { snapshot } = ports;
|
||
let serial: Promise<unknown> = Promise.resolve();
|
||
function exclusive<T>(action: () => Promise<T>): Promise<T> {
|
||
const next = serial.then(action); serial = next.catch(() => undefined); return next;
|
||
}
|
||
async function save(message: string, status: CanvasAgentRunStatus = "running") {
|
||
snapshot.message = message;
|
||
await ports.checkpoint(structuredClone(snapshot), status); return status;
|
||
}
|
||
let modelError = "";
|
||
let verifiedCompletion = "";
|
||
async function finishVerifiedCutout() {
|
||
snapshot.nativeRuntime!.state = "stopped";
|
||
snapshot.messages.push({ role: "assistant", content: verifiedCompletion });
|
||
return save(verifiedCompletion, "completed");
|
||
}
|
||
try {
|
||
if (snapshot.calls.some(call => call.status === "started" || call.uncertain))
|
||
return await save("存在结果未确认的工具调用，请先核对画布，避免重复生成。", "needs_review");
|
||
snapshot.nativeRuntime = { ...snapshot.nativeRuntime, engine: "zcode", state: "starting" };
|
||
const schemas = new Map(Object.entries(ports.definitions).map(([name, definition]) => [name, asSchema(definition.inputSchema!)]));
|
||
const toolDescriptors = Object.entries(ports.tools).map(([name, entry]) => ({ name,
|
||
description: ports.definitions[name]?.description || entry.title,
|
||
inputSchema: schemas.get(name)!.jsonSchema, readOnly: entry.readOnly }));
|
||
toolDescriptors.push({ name: "requestCanvasUserInput", description: "需要用户补充必要信息时提出问题，保存任务并等待用户回答。不要用文件、shell 或虚构数据代替用户输入。",
|
||
inputSchema: { type: "object", properties: { question: { type: "string", minLength: 1, maxLength: 1000 } }, required: ["question"], additionalProperties: false }, readOnly: true });
|
||
let question = "";
|
||
const tools = Object.keys(ports.tools);
|
||
const instructions = [ports.instructions,
|
||
"你正在处理本站 Agent 任务。本站已授权的图片、节点、界面及项目业务能力通过 canvas MCP 提供；视频画布暂未授权，不得绕过限制；先读实际画布，不猜测节点 ID。",
|
||
`可用业务工具：${tools.join("、")}。工具名称带 mcp__canvas__ 前缀。`,
|
||
"原生文件、Bash、记忆及子任务能力仅作用于此任务的私有工作区，不能通过 shell 直接访问模型服务商或本站后台；业务操作使用 MCP。",
|
||
"用户只要求生图或改图时，把结果写入画布并验收即可。除非用户明确要求下载或导出，不要主动调用 image_download，也不要为收尾打开交互面板。",
|
||
"imagePreferences 是用户在输入框选择的生图模型和图片参数，与当前规划模型不同。生成和编辑图片优先用 generateCanvasImages；用户选定的模型不得被技能默认值、你选的其他模型或重试覆盖。自动选择生图模型时，图片工具内部会在明确失败后切换备用模型并补齐结果，不要自行重放付费操作；用户明确指定模型时禁止故障切换。比例、尺寸、数量、反向提示词等控件必须传入；不要通过其他生成接口绕过用户选择。分析、问答、排版等不需要出图的任务无需生成图片。",
|
||
"转正面、换角度、旋转主体视角时，直接用 generateCanvasImages，referenceNodeIds 指向要编辑的真实图片，提示词描述目标视角并保留产品结构和未要求修改的细节。旧多视角功能、angleEditImage 和 image.multi_angle 已废弃，即使历史对话用过也不得再调用或打开旧面板。",
|
||
"查询项目级能力先用 listProjectServices；技能先 loadProjectSkill。项目工具的任务受理、文件选择与后台缺失不等于处理完成。",
|
||
"需要澄清时调用 requestCanvasUserInput；不要使用终端交互。工具返回的 userUpdates 是用户的新补充，必须从下一步纳入。",
|
||
"工具错误涉及结果未确认时停止收费动作，不通过更换调用 ID 或另一个工具重复生成。用户明确要求修改的目标才能删除或覆盖。",
|
||
"面向用户统一称为 Agent。回复只包含用户需要的处理进度和结果，不展示内部引擎品牌、沙箱、执行进程、系统路径、内部配置或原始调试信息。",
|
||
"只有真实工具返回成功才可声称完成；打开面板、等待用户选文件、后台缺少处理服务都不能声称媒体已处理。",
|
||
"单纯抠图去背景的透明性由服务端读取原文件 Alpha 通道核验，预览黑底或棋盘格不是失败证据。工具成功且文件核验通过后直接结束，不重复识图或重新抠图。",
|
||
].join("\n");
|
||
let prompt = snapshot.nativeRuntime.sessionId
|
||
? ["继续当前任务，保留所有已完成结果。", `本轮用户的图片设置：${JSON.stringify(snapshot.imagePreferences || {})}，后续生图以此为准。`, ...snapshot.inputs.filter(input => input.applied).slice(-4).map(input => input.text)].join("\n")
|
||
: `${snapshot.objective}\n\n当前画布与历史对话上下文：${JSON.stringify(ports.context)}`;
|
||
while (true) {
|
||
const control = await exclusive(() => ports.consumeInputs(snapshot));
|
||
if (control.cancelled || ports.signal.aborted) return await save("已停止后续处理，完成的结果保留在画布。", "cancelled");
|
||
await save("正在准备处理任务");
|
||
const startedInputs = new Set(snapshot.inputs.map(input => input.id));
|
||
const attemptController = new AbortController();
|
||
const result = await (ports.native || runNativeZcode)({
|
||
runId: ports.runId, attemptId: crypto.randomUUID(),
|
||
runtimeDir: process.env.CANVAS_ZCODE_RUNTIME_DIR || "/opt/yedir-canvas-zcode/current",
|
||
stateDir: process.env.CANVAS_ZCODE_STATE_DIR || "/var/lib/yedir-canvas-agent",
|
||
nodePath: process.env.CANVAS_ZCODE_NODE_PATH || join(process.env.CANVAS_ZCODE_RUNTIME_DIR || "/opt/yedir-canvas-zcode/current", "bin/node"),
|
||
modelId: ports.gateway.modelId, apiType: ports.gateway.apiType,
|
||
instructions, prompt, context: ports.context, sessionId: snapshot.nativeRuntime.sessionId,
|
||
signal: AbortSignal.any([ports.signal, attemptController.signal]), tools: toolDescriptors, formatToolError: publicAgentError,
|
||
model: async (body, path, signal) => {
|
||
const response = await ports.gateway.forward(body, path, signal);
|
||
if (!response.ok) modelError = response.status === 401 ? "当前 Agent 模型服务商鉴权失败，请检查模型配置"
|
||
: response.status === 429 ? "Agent 模型通道额度不足或请求过多" : "Agent 模型连接中断，已保留执行记录";
|
||
return response;
|
||
},
|
||
onSandbox: async metadata => exclusive(async () => {
|
||
snapshot.nativeRuntime = { ...snapshot.nativeRuntime!, sandbox: metadata.kind, commit: metadata.commit, state: "starting" };
|
||
await save("正在准备处理任务");
|
||
}),
|
||
onEvent: async (event: NativeEvent | NativeResult) => {
|
||
if (!["turn.started", "result", "session.created", "session.resumed"].includes(event.type)) return;
|
||
await exclusive(async () => {
|
||
if (event.sessionId) snapshot.nativeRuntime!.sessionId = event.sessionId;
|
||
if (event.traceId) snapshot.nativeRuntime!.traceId = event.traceId;
|
||
snapshot.nativeRuntime!.state = "running";
|
||
if (event.type === "turn.started") snapshot.steps++;
|
||
await save("正在分析需求并安排处理步骤");
|
||
});
|
||
},
|
||
execute: async (name, input, callId) => {
|
||
if (name === "requestCanvasUserInput") {
|
||
if (typeof input.question !== "string" || !input.question.trim() || input.question.length > 1000) throw new Error("请提供明确的问题");
|
||
question = input.question; return { status: "waiting_for_user", question };
|
||
}
|
||
const entry = ports.tools[name]; if (!entry) throw new Error("此画布工具未授权");
|
||
const schema = schemas.get(name)!;
|
||
const validated = schema.validate ? await schema.validate(input) : { success: true as const, value: input };
|
||
if (!validated.success) throw new Error("画布工具参数无效，未提交业务操作");
|
||
const previous = await exclusive(async () => {
|
||
if (verifiedCompletion) throw new AgentToolRejectedError("任务结果已核验完成，不再执行后续工具");
|
||
if ((await ports.consumeInputs(snapshot)).cancelled || ports.signal.aborted) throw new Error("任务已停止，未提交新操作");
|
||
const old = snapshot.calls.find(call => call.id === callId);
|
||
if (old) {
|
||
if (old.name !== name || JSON.stringify(old.input) !== JSON.stringify(input)) throw new Error("工具调用标识与参数不一致");
|
||
if (old.status === "succeeded") return old.output!;
|
||
throw new Error("此工具结果待确认，禁止自动重复执行");
|
||
}
|
||
if (!entry.readOnly && snapshot.calls.some(call => call.uncertain)) throw new Error("先核对未确认的操作，禁止重复提交收费任务");
|
||
if (!entry.readOnly && !snapshot.calls.some(call => call.name === "planCanvasTask" && call.status === "succeeded"))
|
||
throw new Error("请先制定任务计划再执行画布修改");
|
||
if (!entry.readOnly && snapshot.verification?.retryable && snapshot.verification.failedNodeIds?.length) {
|
||
const targets = [input.nodeId, ...(Array.isArray(input.nodeIds) ? input.nodeIds : []),
|
||
...(Array.isArray(input.referenceNodeIds) ? input.referenceNodeIds : [])].filter((id): id is string => typeof id === "string");
|
||
if (!targets.length || targets.some(id => !snapshot.verification!.failedNodeIds!.includes(id)))
|
||
throw new Error("修复只能处理未通过验收的节点，禁止重复生成已通过的结果");
|
||
}
|
||
snapshot.calls.push({ id: callId, name, title: entry.title, input, status: "started" });
|
||
await save(`正在执行：${entry.title}`); return undefined;
|
||
});
|
||
if (previous) return previous;
|
||
try {
|
||
const output = await entry.execute(validated.value as Record<string, unknown>, callId);
|
||
await exclusive(async () => {
|
||
const call = snapshot.calls.find(call => call.id === callId)!;
|
||
call.status = "succeeded"; call.output = output;
|
||
if (output.projectService && output.status === "succeeded" && typeof output.taskId === "string") {
|
||
for (const accepted of snapshot.calls) if (accepted.output?.projectService && accepted.output.taskId === output.taskId && accepted.output.status === "pending")
|
||
accepted.output = { ...accepted.output, status: "succeeded", verifiedResult: output.result };
|
||
}
|
||
if (name === "planCanvasTask") snapshot.requirements = { ...snapshot.requirements,
|
||
...(typeof output.expectedImageCount === "number" ? { expectedImageCount: output.expectedImageCount } : {}),
|
||
...(typeof output.aspectRatio === "string" ? { aspectRatio: output.aspectRatio } : {}) };
|
||
const ids = Array.isArray(output.createdNodeIds) ? output.createdNodeIds.filter((id): id is string => typeof id === "string") : [];
|
||
snapshot.resultNodeIds = [...new Set([...snapshot.resultNodeIds, ...ids])];
|
||
await save(ids.length ? "图片已返回画布，正在确认结果" : `已完成：${entry.title}`);
|
||
});
|
||
if (name === "removeBackground") {
|
||
// Settle a dedicated cutout only after the actual saved image passes
|
||
// file inspection. Stop this attempt without cancelling the user task.
|
||
await exclusive(async () => {
|
||
if (!isCompletedSimpleCutout(snapshot)) return;
|
||
const inputs = JSON.stringify(snapshot.inputs);
|
||
let verification: AgentRunSnapshot["verification"];
|
||
let control: { cancelled: boolean };
|
||
try { verification = await ports.verify(snapshot); control = await ports.consumeInputs(snapshot); }
|
||
catch { return; } // Leave infrastructure failures for normal review.
|
||
if (control.cancelled || ports.signal.aborted || inputs !== JSON.stringify(snapshot.inputs)) return;
|
||
snapshot.verification = verification;
|
||
if (verification.passed && verification.method === "alpha-channel") {
|
||
verifiedCompletion = "抠图已完成，透明背景 PNG 已放入画布，原图已保留。";
|
||
attemptController.abort();
|
||
}
|
||
});
|
||
}
|
||
return { ...output, userUpdates: snapshot.inputs.filter(input => !startedInputs.has(input.id)).map(input => input.text) };
|
||
} catch (error) {
|
||
await exclusive(async () => {
|
||
const call = snapshot.calls.find(call => call.id === callId)!;
|
||
call.status = "failed"; call.error = publicAgentError(error); call.rejected = error instanceof AgentToolRejectedError; call.uncertain = !entry.readOnly && !call.rejected;
|
||
await save(call.error);
|
||
});
|
||
throw error;
|
||
}
|
||
},
|
||
});
|
||
if (verifiedCompletion && !ports.signal.aborted && !(await ports.isCancelled())) return finishVerifiedCutout();
|
||
await exclusive(async () => {
|
||
snapshot.nativeRuntime!.sessionId = result.sessionId;
|
||
snapshot.nativeRuntime!.state = "stopped";
|
||
snapshot.inputTokens += Number(result.usage?.inputTokens || 0);
|
||
snapshot.outputTokens += Number(result.usage?.outputTokens || 0);
|
||
snapshot.messages.push({ role: "assistant", content: result.response || "" });
|
||
await ports.consumeInputs(snapshot);
|
||
});
|
||
if (await ports.isCancelled()) return await save("已停止后续处理，完成的结果保留在画布。", "cancelled");
|
||
// A failed read is not an uncertain paid operation. Independently verify
|
||
// actual outputs below instead of failing a task because an earlier read timed out.
|
||
if (snapshot.calls.some(call => (call.status === "failed" && !call.rejected && !ports.tools[call.name]?.readOnly) || call.uncertain))
|
||
return await save("部分工具结果待确认，已保留执行记录，请核对后继续。", "needs_review");
|
||
const unresolvedRead = snapshot.calls.some((call, index) => call.status === "failed" && !call.rejected &&
|
||
!snapshot.calls.slice(index + 1).some(later => later.name === call.name && later.status === "succeeded" &&
|
||
JSON.stringify(later.input) === JSON.stringify(call.input)));
|
||
// Without a real result to independently inspect, a viewport/UI action
|
||
// cannot turn a failed analysis into a completed task.
|
||
if (unresolvedRead && !snapshot.resultNodeIds.length)
|
||
return await save("图片或资料暂未读取成功，已保留记录，请稍后继续核对。", "needs_review");
|
||
if (question || snapshot.calls.some(call => call.output?.mode === "interactive" && !isAncillaryDownload(snapshot, call)))
|
||
return await save(question || "请在已打开的工具面板中完成操作，然后继续任务。", "needs_review");
|
||
const supplements = snapshot.inputs.filter(input => !startedInputs.has(input.id));
|
||
if (supplements.length) { prompt = supplements.map(input => input.text).join("\n"); continue; }
|
||
const projectWrites = snapshot.calls.filter(call => !call.rejected && !ports.tools[call.name]?.readOnly && call.output?.projectService);
|
||
const canvasWrites = snapshot.calls.filter(call => !call.rejected && !ports.tools[call.name]?.readOnly && !call.output?.projectService);
|
||
if (projectWrites.length && !canvasWrites.length && !snapshot.requirements?.expectedImageCount) {
|
||
const pending = projectWrites.some(call => call.output?.status !== "succeeded");
|
||
return await save(pending ? "业务任务已受理或等待核对，请读取实际任务状态，保留原任务编号继续。" : result.response || "项目业务操作已完成。", pending ? "needs_review" : "completed");
|
||
}
|
||
// Text/analysis tasks have no generated media to inspect. Mutating tasks
|
||
// still use the existing independent checks against actual saved nodes.
|
||
const mutated = snapshot.calls.some(call => !call.rejected && !ports.tools[call.name]?.readOnly);
|
||
if (!mutated && !snapshot.requirements?.expectedImageCount) {
|
||
const expectsChange = /生成|抠图|去背景|换背景|删除|移除|替换|改名|移动|放大|裁剪|旋转|导出|下载|制作|添加|上传/.test(snapshot.objective)
|
||
&& !/只(?:要|需)?(?:分析|讨论|建议|提示词)|不要(?:执行|生成|修改)/.test(snapshot.objective);
|
||
return await save(expectsChange ? "尚未执行画布操作，请核对任务计划后继续。" : result.response || "已完成画布分析。", expectsChange ? "needs_review" : "completed");
|
||
}
|
||
await save("正在核对画布中的实际结果");
|
||
const reviewSnapshot = { ...snapshot, calls: snapshot.calls.filter(call => !call.rejected &&
|
||
!(call.status === "failed" && ports.tools[call.name]?.readOnly)) };
|
||
const verification = await ports.verify(reviewSnapshot); snapshot.verification = verification;
|
||
snapshot.inputTokens = reviewSnapshot.inputTokens; snapshot.outputTokens = reviewSnapshot.outputTokens;
|
||
if (verification.passed) return await save(result.response || "任务完成，结果已写入画布。", "completed");
|
||
if (verification.retryable && snapshot.repairs < 2) {
|
||
snapshot.repairs++;
|
||
prompt = `验收需要定点修复：${JSON.stringify(verification)}。只处理失败结果，禁止整批重新生成；保留所有通过结果。`;
|
||
continue;
|
||
}
|
||
return await save(`结果待核对：${verification.issues.join("；")}`, "needs_review");
|
||
}
|
||
} catch (error) {
|
||
if (snapshot.nativeRuntime) snapshot.nativeRuntime.state = "stopped";
|
||
const cancelled = await ports.isCancelled().catch(() => false);
|
||
if (verifiedCompletion && !cancelled && !ports.signal.aborted) return finishVerifiedCutout();
|
||
return await save(cancelled ? "已停止后续步骤，完成的结果保留在画布。" : modelError || publicAgentError(error), cancelled ? "cancelled" : "needs_review");
|
||
}
|
||
}
|