import { expect, test } from "@playwright/test"; import { readFileSync } from "node:fs"; import postgres from "postgres"; import { newAgentRunSnapshot } from "@/lib/ai/canvas-agent-runtime/protocol"; import * as store from "@/lib/ai/canvas-agent-runtime/store"; const url = process.env.CANVAS_AGENT_TEST_POSTGRES_URL; test.describe("canvas agent durable run isolation", () => { test.describe.configure({ mode: "serial" }); test.skip(!url, "Uses an explicitly provided disposable PostgreSQL database only"); let client: ReturnType; const owner = crypto.randomUUID(), other = crypto.randomUUID(); const chatId = crypto.randomUUID(), otherChat = crypto.randomUUID(); const projectId = crypto.randomUUID(), otherProject = crypto.randomUUID(); let run: Awaited>; const request = () => ({ id: crypto.randomUUID(), userId: owner, chatId, projectId, assistantMessageId: crypto.randomUUID(), snapshot: newAgentRunSnapshot("先抠图再换背景", [ { role: "user" as const, content: "先抠图再换背景" }, ]), }); test.beforeAll(async () => { if (!url) return; const parsed = new URL(url); if ( parsed.pathname !== "/canvas_agent_runtime_test" || !["127.0.0.1", "localhost"].includes(parsed.hostname) ) { throw new Error("Store tests require a disposable local canvas_agent_runtime_test database"); } if (process.env.POSTGRES_URL !== url) throw new Error("POSTGRES_URL must match the explicitly provided disposable test database"); client = postgres(url, { max: 2 }); await client.unsafe( 'DROP TABLE IF EXISTS "CanvasAgentRun", "CanvasProject", "Chat", "User" CASCADE;', ); await client.unsafe( 'CREATE TABLE "User" ("id" uuid PRIMARY KEY); CREATE TABLE "Chat" ("id" uuid PRIMARY KEY, "userId" uuid NOT NULL); CREATE TABLE "CanvasProject" ("id" uuid PRIMARY KEY, "userId" uuid NOT NULL, "workspaceType" varchar DEFAULT \'canvas\', "canvasData" jsonb, "updatedAt" timestamp DEFAULT now());', ); await client.unsafe(readFileSync("lib/db/migrations/0104_add_canvas_agent_runs.sql", "utf8")); await client`INSERT INTO "User" VALUES (${owner}), (${other})`; await client`INSERT INTO "Chat" VALUES (${chatId}, ${owner}), (${otherChat}, ${other})`; await client`INSERT INTO "CanvasProject" ("id", "userId") VALUES (${projectId}, ${owner}), (${otherProject}, ${other})`; }); test.beforeEach(async () => { await client`DELETE FROM "CanvasAgentRun"`; run = await store.acquireAgentRun(request()); }); test.afterAll(async () => { if (client) await client.end(); }); test("run lookup and controls cannot cross user, chat or project boundaries", async () => { expect(await store.getAgentRun(other, chatId, projectId, run.id)).toBeNull(); expect(await store.getAgentRun(owner, otherChat, projectId, run.id)).toBeNull(); expect(await store.getAgentRun(owner, chatId, otherProject, run.id)).toBeNull(); await expect( store.controlAgentRun({ userId: other, chatId, projectId, runId: run.id, action: "cancel" }), ).rejects.toMatchObject({ status: 404 }); await expect(store.assertAgentRunOwnership(owner, otherProject, chatId)).rejects.toMatchObject({ status: 404, }); expect((await store.getAgentRun(owner, chatId, projectId, run.id))?.cancelRequested).toBe( false, ); }); test("two concurrent submissions acquire at most one live run for the same chat", async () => { await client`DELETE FROM "CanvasAgentRun"`; const attempts = await Promise.allSettled([ store.acquireAgentRun(request()), store.acquireAgentRun(request()), ]); expect(attempts.filter((result) => result.status === "fulfilled")).toHaveLength(1); expect(await client`SELECT id FROM "CanvasAgentRun" WHERE status = 'running'`).toHaveLength(1); }); test("steering input is idempotent and consumed atomically into model history", async () => { const input = { userId: owner, chatId, projectId, runId: run.id, action: "steer" as const, inputId: crypto.randomUUID(), text: "主体颜色保持不变", }; await Promise.all([store.controlAgentRun(input), store.controlAgentRun(input)]); const state = run.snapshot; const consumed = await store.consumeAgentRunInputs(run, state); expect(consumed.inputs).toHaveLength(1); expect(state.messages.at(-1)).toEqual({ role: "user", content: input.text }); expect((await store.consumeAgentRunInputs(run, state)).inputs).toEqual([]); await store.controlAgentRun(input); expect((await store.getAgentRun(owner, chatId, projectId))?.pendingInputs).toEqual([]); }); test("completion cannot race past an accepted steering input or cancellation", async () => { await store.controlAgentRun({ userId: owner, chatId, projectId, runId: run.id, action: "steer", inputId: crypto.randomUUID(), text: "再做一版", }); await expect(store.checkpointAgentRun(run, run.snapshot, "completed")).rejects.toThrow( "完成前收到", ); expect((await store.getAgentRun(owner, chatId, projectId))?.status).toBe("running"); await store.consumeAgentRunInputs(run, run.snapshot); await store.controlAgentRun({ userId: owner, chatId, projectId, runId: run.id, action: "cancel", }); await expect(store.checkpointAgentRun(run, run.snapshot, "completed")).rejects.toThrow( "完成前收到", ); expect((await store.consumeAgentRunInputs(run, run.snapshot)).cancelled).toBe(true); }); test("a resumed run rotates its lease so an old worker cannot overwrite results", async () => { await client`UPDATE "CanvasAgentRun" SET "updatedAt" = ${new Date(Date.now() - 180_000).toISOString()}::timestamp WHERE id = ${run.id}`; const resumed = await store.acquireAgentRun({ ...request(), resumeRunId: run.id }); expect(resumed.leaseId).not.toBe(run.leaseId); await expect(store.checkpointAgentRun(run, run.snapshot)).rejects.toThrow("执行权已变更"); expect(await store.heartbeatAgentRun(run)).toBeNull(); expect((await store.checkpointAgentRun(resumed, resumed.snapshot, "completed")).status).toBe( "completed", ); }); test("an uncertain paid request blocks restart and exposes no resume button", async () => { run.snapshot.calls.push({ id: "billable", name: "generateCanvasImages", input: {}, status: "started", }); await store.checkpointAgentRun(run, run.snapshot); await client`UPDATE "CanvasAgentRun" SET "updatedAt" = ${new Date(Date.now() - 180_000).toISOString()}::timestamp WHERE id = ${run.id}`; await expect(store.acquireAgentRun({ ...request(), resumeRunId: run.id })).rejects.toThrow( "执行结果尚未确认", ); const saved = await store.getAgentRun(owner, chatId, projectId); expect(store.agentRunView(saved!).resumable).toBe(false); expect(store.agentRunView(saved!).status).toBe("interrupted"); }); test("a new task revokes an expired worker instead of reviving two tasks in the same chat", async () => { await client`UPDATE "CanvasAgentRun" SET "updatedAt" = ${new Date(Date.now() - 180_000).toISOString()}::timestamp WHERE id = ${run.id}`; const replacement = await store.acquireAgentRun(request()); expect(replacement.id).not.toBe(run.id); expect(await store.heartbeatAgentRun(run)).toBeNull(); await expect(store.checkpointAgentRun(run, run.snapshot)).rejects.toThrow("执行权已变更"); expect((await store.getAgentRun(owner, chatId, projectId, run.id))?.status).toBe("interrupted"); }); test("video steering updates the next shot settings and keeps the execution ledger", async () => { const state = run.snapshot; state.workspace = "video"; state.videoSettings = { mode: "first-frame", duration: 5, ratio: "16:9", resolution: "720p" }; state.videoRequirements = { expectsVideo: true, expectsScript: true, expectedVideoCount: 1 }; await store.checkpointAgentRun(run, state, "running"); await store.controlAgentRun({ userId: owner, chatId, projectId, runId: run.id, action: "steer", inputId: crypto.randomUUID(), text: "再生成两条视频,改成8秒,9:16", }); await store.consumeAgentRunInputs(run, state); expect(state.videoSettings).toMatchObject({ duration: 8, ratio: "9:16", mode: "first-frame" }); expect(state.videoRequirements?.expectedVideoCount).toBe(3); expect(state.workspace).toBe("video"); }); test("video node appends preserve originals and edges under concurrent writes", async () => { const videoStore = await import("@/lib/ai/canvas-agent-runtime/video-store"); const { serializeVideoRuntimeNode } = await import( "@/lib/ai/canvas-agent-runtime/video-protocol" ); const source = { id: "source", kind: "media", title: "原图", x: 12, y: 34, width: 100, height: 100, accent: "#fff", inputType: "image" as const, draftValues: { mediaUrl: "https://example.test/a.png" }, }; const data = { nodes: [serializeVideoRuntimeNode(source)], edges: [], viewport: { x: 100, y: 200, zoom: 0.7 }, }; await client`UPDATE "CanvasProject" SET "workspaceType" = 'video', "canvasData" = ${client.json(JSON.parse(JSON.stringify(data)))} WHERE id = ${projectId}`; await Promise.all([ videoStore.appendVideoAgentNode(owner, projectId, { ...source, id: "shot-a" }, ["source"]), videoStore.appendVideoAgentNode(owner, projectId, { ...source, id: "shot-b" }, ["source"]), ]); await videoStore.appendVideoAgentNode(owner, projectId, { ...source, id: "shot-a" }, [ "source", ]); const saved = await videoStore.readOwnedVideoCanvas(owner, projectId); expect(saved.nodes).toHaveLength(3); expect(saved.nodes[0]).toEqual(source); expect(saved.project.canvasData?.edges).toHaveLength(2); expect(saved.project.canvasData?.viewport).toEqual(data.viewport); await expect(videoStore.readOwnedVideoCanvas(other, projectId)).rejects.toMatchObject({ status: 404, }); await expect( videoStore.appendVideoAgentNode(other, projectId, { ...source, id: "foreign" }, []), ).rejects.toMatchObject({ status: 404 }); await videoStore.replaceVideoAgentUrl( owner, projectId, "shot-a", "https://example.test/a.png", "https://cdn.example.test/a.png", ); const archived = await videoStore.readOwnedVideoCanvas(owner, projectId); expect(archived.nodes.find((node) => node.id === "shot-a")?.draftValues?.mediaUrl).toBe( "https://cdn.example.test/a.png", ); expect(archived.nodes[0].draftValues?.mediaUrl).toBe("https://example.test/a.png"); await expect( videoStore.saveVideoAgentCanvas( owner, projectId, archived.project.updatedAt.toISOString(), archived.project.canvasData!, ), ).rejects.toThrow("还在执行"); await store.checkpointAgentRun(run, run.snapshot, "needs_review"); await expect( videoStore.saveVideoAgentCanvas( owner, projectId, new Date(0).toISOString(), archived.project.canvasData!, ), ).rejects.toThrow("已有新结果"); await Promise.allSettled([ videoStore.saveVideoAgentCanvas( owner, projectId, archived.project.updatedAt.toISOString(), archived.project.canvasData!, ), videoStore.appendVideoAgentNode(owner, projectId, { ...source, id: "shot-c" }, ["source"]), ]); const final = await videoStore.readOwnedVideoCanvas(owner, projectId); expect(final.nodes.some((node) => node.id === "shot-c")).toBe(true); await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-rename", { action: "rename", nodeIds: ["shot-a"], text: "已改名", }); await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-rename", { action: "rename", nodeIds: ["shot-a"], text: "重复请求不得再执行", }); const renamed = await videoStore.readOwnedVideoCanvas(owner, projectId); expect(renamed.nodes.find((node) => node.id === "shot-a")?.title).toBe("已改名"); const duplicate = await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-copy", { action: "duplicate", nodeIds: ["source"], }); expect(duplicate.createdNodeIds).toHaveLength(1); await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-connect", { action: "connect", nodeIds: ["source", "shot-c"], }); await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-disconnect", { action: "disconnect", nodeIds: ["source", "shot-c"], }); await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-delete", { action: "delete", nodeIds: ["shot-c"], }); const text = await videoStore.mutateVideoAgentCanvas(owner, projectId, "op-text", { action: "create-text", nodeIds: [], text: "可直接显示的文本", }); const edited = await videoStore.readOwnedVideoCanvas(owner, projectId); expect(edited.nodes.find((node) => node.id === text.createdNodeIds[0])).toMatchObject({ kind: "script", inputType: "text", draftValues: { action: "TEXT_RESOURCE", nodeType: "text", scriptRole: "resource", content: ["可直接显示的文本"], }, }); expect(edited.nodes.some((node) => node.id === "shot-c")).toBe(false); expect(edited.nodes.some((node) => node.id === "source")).toBe(true); expect( edited.project.canvasData?.edges?.some( (edge) => edge.source === "shot-c" || edge.target === "shot-c", ), ).toBe(false); await expect( videoStore.mutateVideoAgentCanvas(other, projectId, "foreign-delete", { action: "delete", nodeIds: ["source"], }), ).rejects.toMatchObject({ status: 404 }); }); test("an image run cannot be resumed through the video adapter", async () => { await store.checkpointAgentRun(run, run.snapshot, "needs_review"); const params = request(); params.snapshot.workspace = "video"; await expect(store.acquireAgentRun({ ...params, resumeRunId: run.id })).rejects.toMatchObject({ status: 400, }); }); });