UHD_Media_Tool/apps/server/test/e2e/e4-crash-recovery.test.ts

import { type ChildProcess, spawn } from "node:child_process";
import { fileURLToPath } from "node:url";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { torrentFor } from "./fixtures/modules.ts";
import { crashScript } from "./fixtures/scripts.ts";
import {
  createE2E,
  type E2E,
  PG_URL,
  pgAvailable,
  runEvents,
  runRow,
  seedBase,
  seedMovieTarget,
  waitFor,
} from "./harness.ts";

// E4 崩溃恢复:worker 子进程在 Agent 调用 download.submit 时被 kill -9;新 worker 在租约到期(约 30s)后接手:
// runAgent 的 drive 步骤重跑 → 打开同一 JSONL 会话(pendingOperations>0)→ 按幂等键对账账本里停在 running 的调用
// → resume(模型拿到"结果未知")→ 同参数再调 download.submit 得到账本结果;mock qB 只收到一次 add。

const hasPg = await pgAvailable();
const WORKER = fileURLToPath(new URL("./fixtures/worker.ts", import.meta.url));
const SERVER_DIR = fileURLToPath(new URL("../..", import.meta.url));
const TITLE = "Dune.Part.Two.2024.1080p.BluRay.x264-GRP";

describe.skipIf(!hasPg)("E4 崩溃恢复(worker 子进程 kill -9)", () => {
  let e: E2E;
  const children: ChildProcess[] = [];

  const spawnWorker = (): Promise<ChildProcess> => {
    const child = spawn(process.execPath, ["run", "-A", WORKER], {
      cwd: SERVER_DIR,
      env: {
        ...process.env,
        DATABASE_URL: PG_URL,
        BIZ_SCHEMA: e.biz,
        OW_SCHEMA: e.ow,
        SESSION_DIR: e.sessionDir,
      },
      stdio: ["ignore", "pipe", "pipe"],
    });
    children.push(child);
    let err = "";
    child.stderr!.on("data", (d) => {
      err += d;
    });
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => reject(new Error(`worker 启动超时: ${err}`)), 60_000);
      child.stdout!.on("data", (d) => {
        if (String(d).includes("worker ready")) {
          clearTimeout(timer);
          resolve(child);
        }
      });
      child.on("exit", (code) => {
        clearTimeout(timer);
        reject(new Error(`worker 提前退出 code=${code}: ${err}`));
      });
    });
  };

  beforeAll(async () => {
    e = await createE2E("e4", { worker: false, fakeScripts: [crashScript] });
    const base = await seedBase(e);
    await seedMovieTarget(e, base.siteId);
    await e.db
      .insertInto("automation_grant")
      .values({ action: "download.submit", scope: JSON.stringify({}), granted_by: "user:e2e" })
      .execute();
  }, 120_000);

  afterAll(async () => {
    for (const c of children) if (c.exitCode === null && c.signalCode === null) c.kill("SIGKILL");
    await e?.close();
  }, 60_000);

  it("download.submit 途中 kill worker → 新 worker resume,不重复提交,最终入库", async () => {
    const w1 = await spawnWorker();
    const r = await e.api("POST", "/agent/runs", { goal: "把《沙丘2》1080p 收进来", mode: "execute" });
    expect(r.status).toBe(201);
    const runId = r.body.data.id as string;

    // Agent 调到 download.submit(账本行出现)的瞬间 kill -9
    await waitFor(
      async () =>
        e.db
          .selectFrom("capability_invocation")
          .select("id")
          .where("agent_run_id", "=", runId)
          .where("capability_id", "=", "download.submit")
          .executeTakeFirst(),
      60_000,
      "download.submit 开始",
      10,
    );
    w1.kill("SIGKILL");
    await new Promise((res) => w1.once("exit", res));
    const before = await runRow(e, runId);
    expect(["running", "queued"]).toContain(before.status);

    await spawnWorker();
    // 新 worker 接手(租约约 30s 到期)后提交进入 qB,外部下载器完成
    const hash = torrentFor(TITLE).hash;
    await waitFor(async () => e.qb.get(hash), 120_000, "qB 收到种子", 500);
    await e.qb.complete(hash, { writeFiles: true });
    const done = await waitFor(
      async () => {
        const x = await runRow(e, runId);
        return x.status === "succeeded" ? x : null;
      },
      120_000,
      "run succeeded",
      500,
    );
    expect(done.summary).toContain("已入库 1 个");

    // 不重复提交:qB 只收到一次 add;download.submit 账本只有一条(同幂等键)
    expect(e.qb.callsTo("torrents/add")).toHaveLength(1);
    const inv = await e.db
      .selectFrom("capability_invocation")
      .selectAll()
      .where("agent_run_id", "=", runId)
      .where("capability_id", "=", "download.submit")
      .execute();
    expect(inv).toHaveLength(1);
    expect(inv[0]!.status).toBe("succeeded");
    expect(await e.db.selectFrom("download_task").select("id").execute()).toHaveLength(1);
    // 事件:续跑标记 + 恢复的工具调用
    const events = await runEvents(e, runId);
    expect(events.some((x) => x.type === "run_started" && (x.payload as { resumed: boolean }).resumed)).toBe(true);
    const recovered = events.filter((x) => x.type === "tool_call_end" && (x.payload as { recovery: boolean }).recovery);
    expect(recovered.map((x) => (x.payload as { capabilityId: string }).capabilityId)).toContain("download.submit");
    // (run_id, seq) 连续不冲突
    const seqs = events.map((x) => Number(x.seq));
    expect(new Set(seqs).size).toBe(seqs.length);
    const starts = events
      .filter((x) => x.type === "tool_call_start" && !(x.payload as { recovery: boolean }).recovery)
      .map((x) => (x.payload as { capabilityId: string }).capabilityId);
    expect(starts.slice(-2)).toEqual(["download.submit", "task.inspect"]);
  }, 300_000);
});