import { sql } from "@umt/db"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { DEV_KEY, json } from "../helpers.ts"; import { bearer, createHarness, type Harness, pgAvailable } from "./harness.ts"; // 引擎面板 retry(验收 S8):仅 admin;按原任务名与信封里的原始输入重新入队,幂等键 retry::<序号>; // 运行中/待执行的 run 不能重跑(409);未知 run 404;写审计。 const hasPg = await pgAvailable(); describe.skipIf(!hasPg)("POST /engine/runs/{id}/retry", () => { let h: Harness; let admin: ReturnType; let n = 0; const enqueue = vi.fn(async (_task: string, _input: unknown, _opts: { idempotencyKey: string }) => { n++; await sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) values (${`new-${n}`}, ${_task}, 1, ${_opts.idempotencyKey})`.execute(h.db); return { runId: `new-${n}`, deduplicated: false }; }); beforeAll(async () => { h = await createHarness({ engine: true }); admin = bearer(h.app, DEV_KEY); (h.ctx.engine as unknown as { enqueue: typeof enqueue }).enqueue = enqueue; const t = sql.raw(`"${h.engineSchema}".workflow_runs`); const run = (id: string, status: string, input: unknown) => sql`insert into ${t} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, created_at, updated_at) values ('default', ${id}, 'verifyAndImport', 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 3, now(), now())`.execute(h.db); // 信封里的原始输入含敏感字段:重跑用原值(详情展示才脱敏) await run("f1", "failed", { input: { downloadTaskId: "dt-1", password: "p@ss" }, traceId: "trace-1", priority: 5 }); await run("p1", "pending", { input: { downloadTaskId: "dt-2" }, traceId: "trace-2", priority: 0 }); }, 60_000); afterAll(async () => { await h?.close(); }); it("工作流多值筛选保留排队核验与独立刮削,单值筛选兼容", async () => { const table = sql.raw(`"${h.engineSchema}".workflow_runs`); await sql`insert into ${table} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, created_at, updated_at) values ('default', 'scrape-queue', 'scrapeMedia', 'v2', 'pending', '{}'::jsonb, null, '{"input":{}}'::jsonb, 0, now(), now())`.execute( h.db, ); try { const both = await json( await admin("GET", "/engine/runs?workflowName=verifyAndImport,scrapeMedia&status=pending"), ); expect(both.data.items.map((run: { id: string }) => run.id).sort()).toEqual(["p1", "scrape-queue"]); const single = await json(await admin("GET", "/engine/runs?workflowName=verifyAndImport&status=pending")); expect(single.data.items.map((run: { id: string }) => run.id)).toEqual(["p1"]); } finally { await sql`delete from ${table} where id = 'scrape-queue'`.execute(h.db); } }); it("失败的 run:按原任务名与原始输入重新入队,序号递增,写审计", async () => { const r1 = await json(await admin("POST", "/engine/runs/f1/retry")); expect(r1.data).toEqual({ runId: "new-1", retryOf: "f1", attempt: 1 }); expect(enqueue).toHaveBeenLastCalledWith( "verifyAndImport", { downloadTaskId: "dt-1", password: "p@ss" }, { idempotencyKey: "retry:f1:1", traceId: "trace-1", priority: 5 }, ); const r2 = await json(await admin("POST", "/engine/runs/f1/retry")); expect(r2.data).toMatchObject({ runId: "new-2", attempt: 2 }); expect(enqueue.mock.calls.at(-1)![2].idempotencyKey).toBe("retry:f1:2"); const a = await sql<{ n: number }>` select count(*)::int as n from audit_event where action = 'engine.run_retry'`.execute(h.db); expect(a.rows[0]!.n).toBe(2); }); it("未结束的 run 409,未知 run 404,非 admin 403", async () => { expect((await admin("POST", "/engine/runs/p1/retry")).status).toBe(409); expect((await admin("POST", "/engine/runs/nope/retry")).status).toBe(404); const created = await json(await admin("POST", "/api-keys", { name: "exec", scopes: ["organize.execute"] })); const other = bearer(h.app, created.data.plaintextKey); expect((await other("POST", "/engine/runs/f1/retry")).status).toBe(403); expect(n).toBe(2); }); });