From 2ad089d68a96bd5ff8570c082c66194a236c346b Mon Sep 17 00:00:00 2001 From: Wenxin Duan Date: Tue, 15 Sep 2026 01:29:13 +0800 Subject: [PATCH] fix: resume transient original-job collection and guard probe input --- packages/harness/src/image-video/media.ts | 49 ++++++++++++++----- packages/harness/src/image-video/service.ts | 9 ++-- .../tests/image-video-boundary.test.ts | 4 ++ .../harness/tests/image-video-probe.test.ts | 14 ++++++ .../tests/image-video-transfer.test.ts | 8 +++ packages/harness/tests/image-video.test.ts | 40 ++++++++++++++- scripts/probe-image-video-mcp.mjs | 25 ++++++++++ 7 files changed, 133 insertions(+), 16 deletions(-) create mode 100644 packages/harness/tests/image-video-probe.test.ts create mode 100644 scripts/probe-image-video-mcp.mjs diff --git a/packages/harness/src/image-video/media.ts b/packages/harness/src/image-video/media.ts index f899cbf..aa02c56 100644 --- a/packages/harness/src/image-video/media.ts +++ b/packages/harness/src/image-video/media.ts @@ -1,4 +1,5 @@ import { request } from "node:https"; +import { lookup } from "node:dns/promises"; import { isIP } from "node:net"; import { readFile, writeFile, rename } from "node:fs/promises"; import { join } from "node:path"; @@ -14,7 +15,17 @@ export async function validateImageVideoUrl(value: string, resolver?: AddressRes try { url = new URL(value); } catch { throw new HarnessError("I2V_URL_DENIED", "Invalid media URL"); } // The storage provider may require a signed query. Only remove it from host validation. const hostOnly = new URL(url); hostOnly.search = ""; - const checked = await validateBaseUrl(hostOnly.href, {}, resolver); + let resolutionFailed = false; + const resolveAddresses: AddressResolver = resolver ?? (async hostname => (await lookup(hostname, { all: true })).map(result => ({ address: result.address, family: result.family as 4 | 6 }))); + let checked; + try { + checked = await validateBaseUrl(hostOnly.href, {}, async hostname => { + try { return await resolveAddresses(hostname); } catch { resolutionFailed = true; throw new Error("DNS lookup failed"); } + }); + } catch (error) { + if (resolutionFailed) throw new HarnessError("I2V_TRANSFER_TRANSIENT", "Media DNS lookup temporarily unavailable; retain the original job"); + throw error; + } const address = checked.addresses[0]; if (!address) throw new HarnessError("I2V_URL_DENIED", "No public media address"); return { url, address }; @@ -24,24 +35,30 @@ export async function validateImageVideoUrl(value: string, resolver?: AddressRes export async function imageVideoTransfer(value: string, options: { bytes?: Buffer; contentType?: string; maxBytes?: number } = {}): Promise { const { url, address } = await validateImageVideoUrl(value); return new Promise((resolve, reject) => { - const error = () => new HarnessError("I2V_TRANSFER_FAILED", "Media transfer failed or exceeded its bounds"); + const transient = () => new HarnessError("I2V_TRANSFER_TRANSIENT", "Media transfer temporarily unavailable; retain the original job"); + const rejected = () => new HarnessError("I2V_TRANSFER_REJECTED", "Media response violates the transfer policy or bounds"); + let failure: HarnessError | undefined; + const stop = (error: HarnessError) => { failure = error; req.destroy(error); }; const req = request(url, { family: isIP(address), method: options.bytes ? "PUT" : "GET", headers: options.bytes ? { "Content-Type": options.contentType ?? "application/octet-stream", "Content-Length": options.bytes.length } : {}, lookup: (_hostname, _options, callback) => callback(null, address, isIP(address)), }, response => { - if (!response.statusCode || response.statusCode < 200 || response.statusCode >= 300) { response.destroy(); req.destroy(error()); return; } + if (!response.statusCode || response.statusCode < 200 || response.statusCode >= 300) { + const retryable = response.statusCode !== undefined && ([401, 403, 404, 408, 410, 429].includes(response.statusCode) || response.statusCode >= 500); + stop(retryable ? transient() : rejected()); response.destroy(); return; + } const limit = options.maxBytes ?? (options.bytes ? 1024 * 1024 : 256 * 1024 * 1024); - if (Number(response.headers["content-length"] ?? 0) > limit) { response.destroy(); req.destroy(error()); return; } - if (!options.bytes && !/^video\/mp4(?:;|$)/i.test(response.headers["content-type"] ?? "")) { response.destroy(); req.destroy(error()); return; } + if (Number(response.headers["content-length"] ?? 0) > limit) { stop(rejected()); response.destroy(); return; } + if (!options.bytes && !/^video\/mp4(?:;|$)/i.test(response.headers["content-type"] ?? "")) { stop(rejected()); response.destroy(); return; } let total = 0; const chunks: Buffer[] = []; - response.on("data", (chunk: Buffer) => { total += chunk.length; if (total > limit) req.destroy(error()); else chunks.push(chunk); }); - response.once("error", () => reject(error())); + response.on("data", (chunk: Buffer) => { total += chunk.length; if (total > limit) stop(rejected()); else chunks.push(chunk); }); + response.once("error", () => { clearTimeout(timer); reject(failure ?? transient()); }); response.once("end", () => { clearTimeout(timer); resolve(Buffer.concat(chunks)); }); }); - const timer = setTimeout(() => req.destroy(error()), 120_000); - req.once("error", () => { clearTimeout(timer); reject(error()); }); + const timer = setTimeout(() => stop(transient()), 120_000); + req.once("error", () => { clearTimeout(timer); reject(failure ?? transient()); }); req.end(options.bytes); }); } @@ -66,10 +83,18 @@ export function createImageVideoCollector(options: { skillRoot: string; python: return async (url: string, directory: string, duration: number) => { const bytes = await imageVideoTransfer(url); if (bytes.length < 16 || bytes.subarray(4, 8).toString("ascii") !== "ftyp") throw new HarnessError("I2V_INVALID_MP4", "Artifact is not an MP4 file"); - const path = join(directory, "candidate.mp4"); await writeFile(path, bytes, { mode: 0o600 }); + const path = join(directory, "candidate.mp4"); await localMediaStorage(writeFile(path, bytes, { mode: 0o600 })); const validation = await verifyImageVideo(path, directory, duration, options); - if (validation.sha256 !== sha256(await readFile(path))) throw new HarnessError("I2V_INVALID_MP4", "Validated media changed"); - await rename(path, join(directory, "video.mp4")); + if (validation.sha256 !== sha256(await localMediaStorage(readFile(path)))) throw new HarnessError("I2V_INVALID_MP4", "Validated media changed"); + await localMediaStorage(rename(path, join(directory, "video.mp4"))); return validation; }; } + +async function localMediaStorage(operation: Promise): Promise { + try { return await operation; } catch (error) { + const code = error instanceof Error && "code" in error ? error.code : undefined; + if (typeof code === "string" && ["EIO", "EBUSY", "EMFILE", "ENFILE", "ENOSPC", "EAGAIN"].includes(code)) throw new HarnessError("I2V_TRANSFER_TRANSIENT", "Local media storage temporarily unavailable; retain the original job"); + throw error; + } +} diff --git a/packages/harness/src/image-video/service.ts b/packages/harness/src/image-video/service.ts index 8ae4b76..e190ac1 100644 --- a/packages/harness/src/image-video/service.ts +++ b/packages/harness/src/image-video/service.ts @@ -80,6 +80,7 @@ export class ImageVideoService { 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); + let collecting = false; try { if (!r.jobId) { const result = await this.options.remote.importStatus(r); @@ -90,14 +91,16 @@ export class ImageVideoService { if (result.state === "succeeded") { if (!result.artifactUrl) throw new Error("No unambiguous MP4 artifact"); r.artifactUrl = result.artifactUrl; r.state = "validating"; await this.save(r); + collecting = true; r.validation = validationSchema.parse(await this.options.collect(result.artifactUrl, join(this.root, id), r.draft.durationSeconds)); if (Math.abs(r.validation.videoSeconds - r.draft.durationSeconds) > 0.002 || Math.abs(r.validation.audioSeconds - r.draft.durationSeconds) > 0.05) throw new Error("Duration mismatch"); r.state = "ready"; delete r.error; } else { r.state = result.state; if (result.state === "failed") r.error = "H3 生成失败；未自动重投。"; } } - } catch { - if (r.state === "validating") r.state = "failed"; - r.error = r.state === "failed" ? "成片技术验证失败：请核实 MP4、精确时长、尺寸、帧率和音轨；未自动重投。" : "状态响应暂不可用或契约不明确；保留原任务，下次只读查询。"; + } catch (error) { + const retryCollection = collecting && error instanceof HarnessError && error.code === "I2V_TRANSFER_TRANSIENT"; + if (collecting && !retryCollection) r.state = "failed"; + r.error = retryCollection ? "成片收取暂时失败；30 秒后查询原任务并重试收取，不会重新生成。" : r.state === "failed" ? "成片技术验证失败：请核实 MP4、精确时长、尺寸、帧率和音轨；未自动重投。" : "状态响应暂不可用或契约不明确；保留原任务，下次只读查询。"; } await this.save(r); return this.public(r); }); diff --git a/packages/harness/tests/image-video-boundary.test.ts b/packages/harness/tests/image-video-boundary.test.ts index 4dc025a..8ffd369 100644 --- a/packages/harness/tests/image-video-boundary.test.ts +++ b/packages/harness/tests/image-video-boundary.test.ts @@ -17,6 +17,10 @@ describe("image video remote boundary", () => { for (const url of ["http://cdn.example/v", "https://user:pass@cdn.example/v", "https://127.0.0.1/v", "https://[::ffff:127.0.0.1]/v"]) await expect(harness.validateImageVideoUrl(url)).rejects.toBeDefined(); await expect(harness.validateImageVideoUrl("https://cdn.example/v", () => Promise.resolve([{ address: "192.168.1.2", family: 4 }]))).rejects.toBeDefined(); }); + it("classifies a temporary DNS lookup failure as retryable without allowing private results", async () => { + await expect(harness.validateImageVideoUrl("https://cdn.example/video", () => Promise.reject(new Error("EAI_AGAIN")))).rejects.toMatchObject({ code: "I2V_TRANSFER_TRANSIENT" }); + await expect(harness.validateImageVideoUrl("https://cdn.example/video", () => Promise.resolve([{ address: "127.0.0.1", family: 4 }]))).rejects.toMatchObject({ code: "BASE_URL_NOT_ALLOWED" }); + }); it("extracts only the documented response envelope and rejects malformed or error responses", () => { expect(harness).toHaveProperty("imageVideoResponse"); expect(harness.imageVideoResponse({ content: [{ type: "text", text: '{"job_id":"j1"}' }] })).toEqual({ job_id: "j1" }); diff --git a/packages/harness/tests/image-video-probe.test.ts b/packages/harness/tests/image-video-probe.test.ts new file mode 100644 index 0000000..792d6c4 --- /dev/null +++ b/packages/harness/tests/image-video-probe.test.ts @@ -0,0 +1,14 @@ +import { spawnSync } from "node:child_process"; +import { resolve } from "node:path"; +import { expect, it } from "vitest"; + +it("rejects a TTY before reading credentials and directs the operator to a controlled pipe", () => { + const script = resolve(import.meta.dirname, "../../../scripts/probe-image-video-mcp.mjs"); + // Instrument only the process boundary; do not provide a token or allow a network call. + const source = `Object.defineProperty(process.stdin,'isTTY',{value:true}); const fakeRead=()=>{throw new Error('UNSAFE_STDIN_READ');}; process.stdin.on=fakeRead; await import(${JSON.stringify(script)});`; + const result = spawnSync(process.execPath, ["--input-type=module", "-e", source], { encoding: "utf8", timeout: 5000 }); + expect(result.status).toBe(1); + expect(result.stderr).toContain("controlled pipe"); + expect(result.stderr).not.toContain("UNSAFE_STDIN_READ"); + expect(result.stdout).not.toContain("hidden stdin"); +}); diff --git a/packages/harness/tests/image-video-transfer.test.ts b/packages/harness/tests/image-video-transfer.test.ts index fcc1098..b715874 100644 --- a/packages/harness/tests/image-video-transfer.test.ts +++ b/packages/harness/tests/image-video-transfer.test.ts @@ -30,4 +30,12 @@ describe("bounded pinned HTTPS media transfer", () => { response.status = 200; response.type = "text/html"; await expect(imageVideoTransfer("https://cdn.example/video")).rejects.toThrow(); response.type = "video/mp4"; await expect(imageVideoTransfer("https://cdn.example/video", { maxBytes: 2 })).rejects.toThrow(); }); + it("marks temporary HTTP download failures retryable but redirect and content failures permanent", async () => { + response.status = 503; response.type = "video/mp4"; + await expect(imageVideoTransfer("https://cdn.example/video")).rejects.toMatchObject({ code: "I2V_TRANSFER_TRANSIENT" }); + response.status = 302; + await expect(imageVideoTransfer("https://cdn.example/video")).rejects.toMatchObject({ code: "I2V_TRANSFER_REJECTED" }); + response.status = 200; response.type = "text/html"; + await expect(imageVideoTransfer("https://cdn.example/video")).rejects.toMatchObject({ code: "I2V_TRANSFER_REJECTED" }); + }); }); diff --git a/packages/harness/tests/image-video.test.ts b/packages/harness/tests/image-video.test.ts index 60c3950..62d5677 100644 --- a/packages/harness/tests/image-video.test.ts +++ b/packages/harness/tests/image-video.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { Readable } from "node:stream"; import sharp from "sharp"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import * as harness from "../src/index.js"; import type { ImageVideoRemote, ImageVideoRecord } from "../src/image-video/contracts.js"; @@ -111,6 +111,44 @@ describe("durable image video tasks", () => { 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); diff --git a/scripts/probe-image-video-mcp.mjs b/scripts/probe-image-video-mcp.mjs new file mode 100644 index 0000000..2852296 --- /dev/null +++ b/scripts/probe-image-video-mcp.mjs @@ -0,0 +1,25 @@ +// Read-only discovery with the application's configured MCP credential. +import {createInterface} from 'node:readline'; +import {createRequire} from 'node:module'; +import {writeFile} from 'node:fs/promises'; +if(process.stdin.isTTY){console.error('Refusing TTY credential input. Provide the token through a controlled pipe from a trusted credential source; do not type it into this terminal or put it in command arguments.');process.exit(1);} +const require=createRequire(new URL('../packages/harness/package.json',import.meta.url)); +const {Client}=require('@modelcontextprotocol/sdk/client/index.js'); +const {StreamableHTTPClientTransport}=require('@modelcontextprotocol/sdk/client/streamableHttp.js'); +const reader=createInterface({input:process.stdin}); +console.log('Reading MCP token from controlled non-TTY stdin; read-only schema and identity discovery.'); +const token=(await reader[Symbol.asyncIterator]().next()).value?.trim(); reader.close(); +if(!token)process.exit(1); +const names=new Set(['video_h3_submit','video_job_status','creative_upload_from_url','creative_import_status','swads_whoami','video_h3_workers']); +const client=new Client({name:'agent-studio-i2v-contract-probe',version:'1.0.0'}); +try { + await client.connect(new StreamableHTTPClientTransport(new URL('https://ads.mincode.cn/mcp'),{requestInit:{redirect:'error',headers:{Authorization:`Bearer ${token}`}}}),{timeout:20000}); + let cursor;const schemas=[]; + do {const page=await client.listTools(cursor?{cursor}:{});schemas.push(...page.tools.filter(tool=>names.has(tool.name)));cursor=page.nextCursor;}while(cursor); + const identity=await client.callTool({name:'swads_whoami',arguments:{}}); + const body=identity.structuredContent??JSON.parse(identity.content.find(item=>item.type==='text').text); + const account=body.access?.current_account; + const report={checkedAt:new Date().toISOString(),toolSchemas:schemas,identityShape:{rootKeys:Object.keys(body),accountPath:'access.current_account',tenant_id:account?.tenant_id,platform:account?.platform,hasGeneratePermission:account?.permissions?.includes('creatives.generate'),hasUploadPermission:account?.permissions?.includes('creatives.upload')},note:'Read-only; no upload/generation or account mutation. Do not bind upload to external account. Output schemas may be absent; do not guess response fields.'}; + await writeFile('.superpowers/sdd/2026-09-15-image-to-video-web/live-contracts.json',JSON.stringify(report,null,2).replaceAll(token,'[REDACTED]'),{mode:0o600}); + console.log(JSON.stringify({found:schemas.map(tool=>({name:tool.name,hasOutputSchema:!!tool.outputSchema,inputRequired:tool.inputSchema.required})),identity:report.identityShape})); +}catch(error){console.error(JSON.stringify({error:String(error.message).replaceAll(token,'[REDACTED]').slice(0,500)}));process.exitCode=1;}finally{await client.close();}