245 lines
19 KiB
TypeScript
245 lines
19 KiB
TypeScript
import { mkdtemp, mkdir, readFile, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { Readable } from "node:stream";
|
|
import sharp from "sharp";
|
|
import { describe, expect, it, vi } from "vitest";
|
|
import * as harness from "../src/index.js";
|
|
import type { ImageVideoRemote, ImageVideoRecord } from "../src/image-video/contracts.js";
|
|
|
|
async function fixture() {
|
|
expect(harness).toHaveProperty("ImageVideoService");
|
|
const root = await mkdtemp(join(tmpdir(), "i2v-test-"));
|
|
await mkdir(join(root, "ses_owner"));
|
|
const attachments = new harness.FileAttachmentStore({ sessionsRoot: root });
|
|
const attachment = await attachments.add("ses_owner", { name: "test.png", mediaType: "image/png", stream: Readable.from(await sharp({ create: { width: 8, height: 8, channels: 3, background: "red" } }).png().toBuffer()) });
|
|
const calls: string[] = [];
|
|
const remote: ImageVideoRemote = {
|
|
preflight() { calls.push("identity"); return Promise.resolve({ tenantId: "tenant", platform: "tiktok" }); },
|
|
initialize() { calls.push("initialize"); return Promise.resolve({ uploadId: "upload_1", objectKey: "obj", uploadUrl: "https://public.example/put?token=private" }); },
|
|
put() { calls.push("put"); return Promise.resolve(); }, finalize() { calls.push("finalize"); return Promise.resolve(); },
|
|
importStatus() { calls.push("import"); return Promise.resolve({ state: "ready", assetId: "asset_1" }); },
|
|
submit() { calls.push("submit"); return Promise.resolve({ jobId: "remote_job_1" }); },
|
|
status() { calls.push("status"); return Promise.resolve({ state: "queued" }); },
|
|
};
|
|
const options = { root: join(root, "tasks"), attachments, remote, collect: () => Promise.reject(new Error("No fake video")) };
|
|
const service = new harness.ImageVideoService(options);
|
|
const draft = { attachmentId: attachment.attachmentId, title: "Motion", prompt: "Natural movement with ambient sound", durationSeconds: 10 };
|
|
return { service, options, draft, calls, remote, root };
|
|
}
|
|
|
|
describe("durable image video tasks", () => {
|
|
it("continues an orphaned uploaded task locally without losing receipts or automatically submitting", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
const path = join(f.options.root, task.id, "record.json");
|
|
const before = JSON.parse(await readFile(path, "utf8")) as ImageVideoRecord;
|
|
f.calls.length = 0;
|
|
const restored = new harness.ImageVideoService(f.options);
|
|
const result = await restored.continueTask("ses_new", task.id, before.revision, () => false, () => {});
|
|
expect(result).toMatchObject({ id: task.id, state: "awaiting_submission", revision: before.revision + 1 });
|
|
expect(f.calls).toEqual([]);
|
|
const after = JSON.parse(await readFile(path, "utf8")) as ImageVideoRecord;
|
|
expect(after).toMatchObject({ ownerSessionId: "ses_new", draft: before.draft, referenceHash: before.referenceHash, requestHash: before.requestHash, upload: before.upload, assetId: "asset_1" });
|
|
expect(after.continuations).toHaveLength(1);
|
|
expect(after.continuations?.[0]).toMatchObject({ fromSessionId: "ses_owner", toSessionId: "ses_new", revision: before.revision });
|
|
expect(Number.isFinite(Date.parse(after.continuations?.[0]?.at ?? ""))).toBe(true);
|
|
await expect(restored.owned("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
|
|
expect((await restored.owned("ses_new", task.id)).id).toBe(task.id);
|
|
await restored.submit("ses_new", task.id);
|
|
expect(f.calls).toEqual(["identity", "submit"]);
|
|
expect(JSON.stringify(result)).not.toMatch(/ownerSessionId|continuations|uploadUrl/);
|
|
});
|
|
it("rejects active-owner takeover, stale confirmation, uncertain state and concurrent continuation", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await expect(f.service.continueTask("ses_new", task.id, 0, () => true, () => {})).rejects.toMatchObject({ code: "I2V_OWNER_ACTIVE" });
|
|
await expect(f.service.continueTask("ses_new", task.id, 99, () => false, () => {})).rejects.toMatchObject({ code: "I2V_STALE_REVISION" });
|
|
const attempts = await Promise.allSettled([f.service.continueTask("ses_new", task.id, 0, () => false, () => {}), f.service.continueTask("ses_other", task.id, 0, () => false, () => {})]);
|
|
expect(attempts.map(a => a.status)).toEqual(["fulfilled", "rejected"]);
|
|
const path = join(f.options.root, task.id, "record.json");
|
|
const record = JSON.parse(await readFile(path, "utf8")) as ImageVideoRecord;
|
|
for (const state of ["unknown_upload", "unknown_submission", "failed", "queued"] as const) {
|
|
await writeFile(path, JSON.stringify({ ...record, state }));
|
|
await expect(f.service.continueTask("ses_new", task.id, record.revision, () => false, () => {})).rejects.toMatchObject({ code: "I2V_CONTINUE_BLOCKED" });
|
|
expect((JSON.parse(await readFile(path, "utf8")) as ImageVideoRecord).state).toBe(state);
|
|
}
|
|
expect(f.calls).toEqual([]);
|
|
});
|
|
it("reuses a verified existing image without PUT, finalize or automatic generation and persists it across restart", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
f.remote.initialize = () => { f.calls.push("initialize"); return Promise.resolve({ reused: true as const, assetId: "existing_asset" }); };
|
|
expect((await f.service.upload("ses_owner", task.id)).state).toBe("awaiting_submission");
|
|
expect(f.calls).toEqual(["identity", "initialize"]);
|
|
const restored = new harness.ImageVideoService(f.options);
|
|
expect((await restored.view(task.id)).state).toBe("awaiting_submission");
|
|
expect((await restored.submit("ses_owner", task.id)).jobId).toBe("remote_job_1");
|
|
expect(f.calls).not.toContain("put"); expect(f.calls).not.toContain("finalize");
|
|
});
|
|
it("persists a sanitized failure stage while keeping rejected uploads quarantined", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
f.remote.initialize = () => Promise.reject(new harness.HarnessError("I2V_REMOTE_REJECTED", "secret signed URL"));
|
|
const result = await f.service.upload("ses_owner", task.id);
|
|
expect(result).toMatchObject({ state: "unknown_upload" });
|
|
expect(result.error).toContain("initializing_upload/I2V_REMOTE_REJECTED");
|
|
expect(result.error).not.toContain("secret");
|
|
expect((await new harness.ImageVideoService(f.options).view(task.id)).error).toBe(result.error);
|
|
expect(f.calls).not.toContain("put");
|
|
});
|
|
it("copies a local reference without transmitting it and hides private bindings", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
expect(f.calls).toEqual([]); expect(task.state).toBe("draft");
|
|
await f.options.attachments.removeAll("ses_owner");
|
|
expect((await f.service.media(task.id, "reference")).size).toBeGreaterThan(0);
|
|
expect(JSON.stringify(await f.service.list())).not.toMatch(/ownerSessionId|uploadUrl|objectKey|referenceHash|\/private\//);
|
|
});
|
|
it("rejects foreign owners before any external operation", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await expect(f.service.upload("ses_other", task.id)).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
|
|
await expect(f.service.submit("ses_other", task.id)).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
|
|
expect(f.calls).toEqual([]);
|
|
});
|
|
it("requires upload first and coalesces concurrent submission, restoring only the known job", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await expect(f.service.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_NOT_READY" });
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
const [a, b] = await Promise.all([f.service.submit("ses_owner", task.id), f.service.submit("ses_owner", task.id)]);
|
|
expect(a.jobId).toBe("remote_job_1"); expect(b.jobId).toBe(a.jobId);
|
|
const restored = new harness.ImageVideoService(f.options);
|
|
expect((await restored.view(task.id)).jobId).toBe(a.jobId);
|
|
await restored.submit("ses_owner", task.id);
|
|
expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
|
|
expect(JSON.stringify(await restored.view(task.id))).not.toContain("private");
|
|
});
|
|
it("isolates unknown submissions after timeout and after process interruption", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
f.remote.submit = () => { f.calls.push("submit"); return Promise.reject(new Error("token=private")); };
|
|
expect((await f.service.submit("ses_owner", task.id)).state).toBe("unknown_submission");
|
|
await expect(new harness.ImageVideoService(f.options).submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" });
|
|
expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
|
|
const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
|
|
record.state = "submitting"; await writeFile(file, JSON.stringify(record));
|
|
expect((await new harness.ImageVideoService(f.options).view(task.id)).state).toBe("unknown_submission");
|
|
});
|
|
it("records a typed submission rejection without exposing provider details or retrying", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
f.remote.submit = () => Promise.reject(new harness.HarnessError("I2V_REMOTE_REJECTED", "secret upstream detail"));
|
|
const result = await f.service.submit("ses_owner", task.id);
|
|
expect(result.error).toContain("submitting/I2V_REMOTE_REJECTED");
|
|
expect(result.error).not.toContain("secret");
|
|
expect(result.state).toBe("unknown_submission");
|
|
await expect(f.service.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" });
|
|
});
|
|
it("never reinitializes uncertain uploads and blocks stale persisted revisions", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
f.remote.initialize = () => { f.calls.push("initialize"); return Promise.reject(new Error("timeout")); };
|
|
expect((await f.service.upload("ses_owner", task.id)).state).toBe("unknown_upload");
|
|
await expect(f.service.upload("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_UPLOAD" });
|
|
expect(f.calls.filter(call => call === "initialize")).toHaveLength(1);
|
|
const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
|
|
record.requestHash = "invalid"; await writeFile(file, JSON.stringify(record));
|
|
await expect(f.service.view(task.id)).rejects.toMatchObject({ code: "I2V_RECORD_INVALID" });
|
|
});
|
|
it("throttles read-only polling even across instances and accepts no arbitrary identifiers", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
|
|
await Promise.all([f.service.refresh(task.id), new harness.ImageVideoService(f.options).refresh(task.id)]);
|
|
expect(f.calls.filter(call => call === "status")).toHaveLength(1);
|
|
await expect(f.service.view("../../etc/passwd")).rejects.toMatchObject({ code: "I2V_TASK_NOT_FOUND" });
|
|
});
|
|
it("rejects a late submit result from overwriting a newer revision, while retaining its known receipt", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
f.remote.submit = async () => {
|
|
const file = join(f.options.root, task.id, "record.json"); const newer = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
|
|
newer.revision += 1; newer.state = "unknown_submission"; await writeFile(file, JSON.stringify(newer));
|
|
return { jobId: "late_job" };
|
|
};
|
|
await expect(f.service.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_STALE_REVISION" });
|
|
expect((await f.service.view(task.id)).state).toBe("unknown_submission");
|
|
const receipt = JSON.parse(await readFile(join(f.options.root, task.id, "submission-receipt.json"), "utf8")) as { jobId: string };
|
|
expect(receipt.jobId).toBe("late_job");
|
|
});
|
|
it("recovers a matching persisted submit receipt after restart and polls that job without resubmitting", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
|
|
record.state = "submitting"; record.revision++;
|
|
await writeFile(file, JSON.stringify(record));
|
|
await writeFile(join(f.options.root, task.id, "submission-receipt.json"), JSON.stringify({ version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "accepted_before_crash" }));
|
|
f.calls.length = 0;
|
|
f.remote.status = (r) => { f.calls.push(`status:${r.jobId ?? "missing"}`); return Promise.resolve({ state: "running" }); };
|
|
const restored = new harness.ImageVideoService(f.options);
|
|
expect(await restored.view(task.id)).toMatchObject({ state: "queued", jobId: "accepted_before_crash" });
|
|
expect(JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord).toMatchObject({ state: "queued", jobId: "accepted_before_crash", revision: record.revision + 1 });
|
|
expect(await restored.refresh(task.id)).toMatchObject({ state: "running", jobId: "accepted_before_crash" });
|
|
await new harness.ImageVideoService(f.options).submit("ses_owner", task.id);
|
|
expect(f.calls).toEqual(["status:accepted_before_crash"]);
|
|
});
|
|
it("retries a transient download after restart using the original job's current artifact, at the polling cadence", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
|
|
const clock = vi.spyOn(Date, "now").mockReturnValue(2_000_000_000_000);
|
|
const downloaded: string[] = []; let queries = 0;
|
|
f.remote.status = r => { f.calls.push(`status:${r.jobId ?? "missing"}`); queries++; return Promise.resolve({ state: "succeeded", artifactUrl: `https://public.example/${String(queries)}.mp4` }); };
|
|
const collect = (url: string) => {
|
|
downloaded.push(url);
|
|
if (downloaded.length === 1) return Promise.reject(new harness.HarnessError("I2V_TRANSFER_TRANSIENT", "Temporary download timeout"));
|
|
return Promise.resolve({ videoSeconds: 10, audioSeconds: 10, width: 720, height: 1280, fps: 24, decoded: true as const, sha256: "a".repeat(64), visualReview: "pending" as const, audioReview: "pending" as const });
|
|
};
|
|
try {
|
|
const initial = new harness.ImageVideoService({ ...f.options, collect });
|
|
expect(await initial.refresh(task.id)).toMatchObject({ state: "validating", jobId: "remote_job_1" });
|
|
const restored = new harness.ImageVideoService({ ...f.options, collect });
|
|
await restored.refresh(task.id); expect(downloaded).toHaveLength(1);
|
|
clock.mockReturnValue(2_000_000_029_999); await restored.refresh(task.id); expect(downloaded).toHaveLength(1);
|
|
clock.mockReturnValue(2_000_000_030_000);
|
|
expect(await restored.refresh(task.id)).toMatchObject({ state: "ready", jobId: "remote_job_1" });
|
|
await restored.submit("ses_owner", task.id);
|
|
expect(downloaded).toEqual(["https://public.example/1.mp4", "https://public.example/2.mp4"]);
|
|
expect(f.calls.filter(call => call.startsWith("status:"))).toEqual(["status:remote_job_1", "status:remote_job_1"]);
|
|
expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
|
|
} finally { clock.mockRestore(); }
|
|
});
|
|
it("keeps deterministic invalid media terminal across restart with no false ready or second generation", async () => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); await f.service.submit("ses_owner", task.id);
|
|
f.remote.status = () => Promise.resolve({ state: "succeeded", artifactUrl: "https://public.example/invalid.mp4" });
|
|
let collections = 0;
|
|
const collect = () => { collections++; return Promise.reject(new harness.HarnessError("I2V_INVALID_MP4", "Invalid MP4")); };
|
|
expect((await new harness.ImageVideoService({ ...f.options, collect }).refresh(task.id)).state).toBe("failed");
|
|
const restored = new harness.ImageVideoService({ ...f.options, collect });
|
|
expect(await restored.refresh(task.id)).toMatchObject({ state: "failed", jobId: "remote_job_1" });
|
|
await restored.submit("ses_owner", task.id);
|
|
expect(collections).toBe(1); expect(f.calls.filter(call => call === "submit")).toHaveLength(1);
|
|
await expect(restored.media(task.id, "video")).rejects.toMatchObject({ code: "I2V_MEDIA_NOT_FOUND" });
|
|
});
|
|
it.each(["task", "hash", "stale", "future", "malformed", "empty_job", "version", "extra_field", "conflicting_job"] as const)("quarantines a %s receipt without polling it or submitting again", async (kind) => {
|
|
const f = await fixture(); const task = await f.service.create("ses_owner", f.draft);
|
|
await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id);
|
|
const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord;
|
|
record.state = "submitting"; record.revision++;
|
|
delete record.lastPollAt;
|
|
if (kind === "conflicting_job") record.jobId = "different_record_job";
|
|
const receipt: Record<string, unknown> = { version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "untrusted_receipt_job" };
|
|
if (kind === "task") receipt.id = "i2v_11111111-1111-1111-1111-111111111111";
|
|
if (kind === "hash") receipt.requestHash = "a".repeat(64);
|
|
if (kind === "stale") receipt.revision = record.revision - 1;
|
|
if (kind === "future") receipt.revision = record.revision + 1;
|
|
if (kind === "empty_job") receipt.jobId = "";
|
|
if (kind === "version") receipt.version = 2;
|
|
if (kind === "extra_field") receipt.unexpected = "conflicting format";
|
|
const serialized = kind === "malformed" ? "{broken-json" : JSON.stringify(receipt);
|
|
await writeFile(file, JSON.stringify(record));
|
|
const receiptPath = join(f.options.root, task.id, "submission-receipt.json"); await writeFile(receiptPath, serialized);
|
|
f.calls.length = 0;
|
|
const restored = new harness.ImageVideoService(f.options);
|
|
expect((await restored.view(task.id)).state).toBe("unknown_submission");
|
|
expect((await restored.refresh(task.id)).state).toBe("unknown_submission");
|
|
await expect(restored.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" });
|
|
expect(f.calls).toEqual([]);
|
|
expect(await readFile(receiptPath, "utf8")).toBe(serialized);
|
|
});
|
|
});
|