wenxinduan50/agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recover matching durable image video submission receipts

Browse Source
This commit is contained in:
Wenxin Duan
2026-09-15 01:14:25 +08:00
parent 40e03e393c
commit cc80265bb2
2 changed files with 59 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/harness/src/image-video/service.ts
+17 -3
View File
@@ -6,6 +6,7 @@ import { HarnessError } from "../errors/harness-error.js";
import { ImageVideoDraftSchema, recordSchema, taskIdSchema, validationSchema, type ImageVideoDraft, type ImageVideoPublicTask, type ImageVideoRecord, type ImageVideoRemote, type ImageVideoValidation } from "./contracts.js";
const locks = new Map<string, Promise<unknown>>();
const submissionReceiptSchema = recordSchema.pick({ version: true, id: true, revision: true, requestHash: true, jobId: true }).required().strict();
export const sha256 = (bytes: Buffer | string) => createHash("sha256").update(bytes).digest("hex");
const fail = (code: string, message: string): never => { throw new HarnessError(code, message); };
export type ImageVideoOptions = { root: string; attachments: Pick<AttachmentStore, "get" | "openNormalized">; remote: ImageVideoRemote; collect: (url: string, directory: string, duration: number) => Promise<ImageVideoValidation> };
@@ -57,8 +58,8 @@ export class ImageVideoService {
async submit(sessionId: string, id: string): Promise<ImageVideoPublicTask> {
return this.lock(id, async () => {
const r = await this.read(id); this.owner(r, sessionId);
if (r.jobId) return this.public(r);
if (r.state === "unknown_submission") return fail("I2V_UNKNOWN_SUBMISSION", "Submission outcome unknown; no safe retry or new billing request is allowed");
if (r.jobId) return this.public(r);
if (r.state !== "awaiting_submission" || !r.assetId) return fail("I2V_NOT_READY", "Reference import must be ready before generation approval");
await this.options.remote.preflight(); await this.reference(r);
r.state = "submitting"; await this.save(r);
@@ -76,7 +77,7 @@ export class ImageVideoService {
async refresh(id: string): Promise<ImageVideoPublicTask> {
return this.lock(id, async () => {
const r = await this.read(id);
if (r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r);
if (r.state === "unknown_submission" || r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r);
if (!r.jobId && !(r.upload && ["importing", "unknown_upload"].includes(r.state))) return this.public(r);
r.lastPollAt = Date.now(); await this.save(r);
try {
@@ -118,6 +119,15 @@ export class ImageVideoService {
}
private async reference(r: ImageVideoRecord) { await this.regular(join(this.root, r.id, "reference")); const bytes = await readFile(join(this.root, r.id, "reference")); if (sha256(bytes) !== r.referenceHash) fail("I2V_MEDIA_INVALID", "Reference changed after draft"); return bytes; }
private async regular(path: string) { const info = await lstat(path); if (!info.isFile() || info.isSymbolicLink()) fail("I2V_RECORD_INVALID", "Unsafe registered file"); }
private async matchingSubmissionReceipt(r: ImageVideoRecord): Promise<string | undefined> {
const path = join(this.root, r.id, "submission-receipt.json");
try {
await this.regular(path);
const receipt = submissionReceiptSchema.parse(JSON.parse(await readFile(path, "utf8")));
if (receipt.id !== r.id || receipt.requestHash !== r.requestHash || receipt.revision !== r.revision || !receipt.jobId.trim() || (r.jobId !== undefined && r.jobId !== receipt.jobId)) return undefined;
return receipt.jobId;
} catch { return undefined; } // Preserve absent, malformed and conflicting evidence for manual reconciliation.
}
private async read(id: string): Promise<ImageVideoRecord> {
if (!taskIdSchema.safeParse(id).success) return fail("I2V_TASK_NOT_FOUND", "Task not found");
const path = join(this.root, id, "record.json");
@@ -126,7 +136,11 @@ export class ImageVideoService {
try {
await this.regular(path); const r = recordSchema.parse(JSON.parse(await readFile(path, "utf8")));
if (r.id !== id || r.requestHash !== sha256(JSON.stringify(r.draft))) throw new Error("Binding mismatch");
if (r.state === "submitting") { r.state = "unknown_submission"; await this.save(r); }
if (r.state === "submitting") {
const jobId = await this.matchingSubmissionReceipt(r);
if (jobId !== undefined) { r.jobId = jobId; r.state = "queued"; delete r.lastPollAt; delete r.error; await this.save(r); }
else { r.state = "unknown_submission"; await this.save(r); }
}
if (["initializing_upload", "uploading", "finalizing"].includes(r.state)) { r.state = "unknown_upload"; await this.save(r); }
return r;
} catch { return fail("I2V_RECORD_INVALID", "Task record is invalid; manual recovery required"); }
packages/harness/tests/image-video.test.ts
+42
View File
@@ -95,4 +95,46 @@ describe("durable image video tasks", () => {
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.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);
});
});