ai-chatbot/tests/routes/canvas-agent-runtime-store.test.ts

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<typeof postgres>;
  const owner = crypto.randomUUID(),
    other = crypto.randomUUID();
  const chatId = crypto.randomUUID(),
    otherChat = crypto.randomUUID();
  const projectId = crypto.randomUUID(),
    otherProject = crypto.randomUUID();
  let run: Awaited<ReturnType<typeof store.acquireAgentRun>>;
  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,
    });
  });
});