import { describe, expect, it, vi } from "vitest"; import { clampPdfToMdConcurrency, invokePdfToMdBatch, mapPool, readPdfToMdConcurrency, } from "../../src/capability/pdfToMdBundle.js"; import type { CapabilityAdapter, CapabilityInvocationResult } from "../../src/capability/types.js"; function okResult(label: string, pages: number): CapabilityInvocationResult { return { artifacts: [{ path: `out/${label}/document.md`, kind: "markdown" }], consumption: { provider: "aliyun_docmind", model: null, inputTokens: null, outputTokens: null, quantity: pages, unit: "pages", costUsd: pages * 0.0056, correlationId: `job-${label}`, }, }; } describe("pdf_to_md batch concurrency helpers", () => { it("mapPool caps in-flight workers", async () => { let inFlight = 0; let maxInFlight = 0; const values = await mapPool([1, 2, 3, 4, 5], 2, async (item) => { inFlight += 1; maxInFlight = Math.max(maxInFlight, inFlight); await new Promise((resolve) => setTimeout(resolve, 40)); inFlight -= 1; return item * 10; }); expect(values).toEqual([10, 20, 30, 40, 50]); expect(maxInFlight).toBe(2); }); it("clamps concurrency and reads env", () => { expect(clampPdfToMdConcurrency(99)).toBe(8); expect(clampPdfToMdConcurrency(0)).toBe(1); expect(readPdfToMdConcurrency({})).toBe(3); expect(readPdfToMdConcurrency({ HUB_PDF_TO_MD_MAX_CONCURRENT: "5" })).toBe(5); expect(() => readPdfToMdConcurrency({ HUB_PDF_TO_MD_MAX_CONCURRENT: "nope" })).toThrow(/positive integer/); }); it("runs batch items concurrently and preserves order", async () => { let inFlight = 0; let maxInFlight = 0; const adapter: CapabilityAdapter = { capabilityId: "pdf_to_md_bundle", invoke: vi.fn(async (input) => { inFlight += 1; maxInFlight = Math.max(maxInFlight, inFlight); await new Promise((resolve) => setTimeout(resolve, 60)); inFlight -= 1; const label = input.inputPath.includes("b") ? "b" : input.inputPath.includes("c") ? "c" : "a"; return okResult(label, label === "a" ? 1 : label === "b" ? 2 : 3); }), }; const batch = await invokePdfToMdBatch( adapter, { runId: "run-1", organizationId: "org-1", projectId: "proj-1", workspaceDir: "/tmp/ws", prisma: {} as never, }, [ { inputPath: "a.pdf", outputDir: "out/a" }, { inputPath: "b.pdf", outputDir: "out/b" }, { inputPath: "c.pdf", outputDir: "out/c" }, ], 3, ); expect(batch.map((item) => item.ok)).toEqual([true, true, true]); expect(maxInFlight).toBe(3); expect(adapter.invoke).toHaveBeenCalledTimes(3); if (batch[0]?.ok && batch[1]?.ok && batch[2]?.ok) { expect(batch[0].result.consumption.correlationId).toBe("job-a"); expect(batch[1].result.consumption.correlationId).toBe("job-b"); expect(batch[2].result.consumption.correlationId).toBe("job-c"); } }); it("isolates per-item failures without canceling siblings", async () => { const adapter: CapabilityAdapter = { capabilityId: "pdf_to_md_bundle", invoke: vi.fn(async (input) => { if (input.inputPath.includes("bad")) { throw new Error("boom"); } return okResult("ok", 1); }), }; const batch = await invokePdfToMdBatch( adapter, { runId: "run-1", organizationId: "org-1", projectId: "proj-1", workspaceDir: "/tmp/ws", prisma: {} as never, }, [ { inputPath: "ok.pdf", outputDir: "out/ok" }, { inputPath: "bad.pdf", outputDir: "out/bad" }, ], 2, ); expect(batch[0]?.ok).toBe(true); expect(batch[1]?.ok).toBe(false); if (batch[1]?.ok === false) { expect(batch[1].error).toMatch(/boom/); } }); it("rejects duplicate output dirs before starting work", async () => { const invoke = vi.fn(); const adapter: CapabilityAdapter = { capabilityId: "pdf_to_md_bundle", invoke, }; await expect(invokePdfToMdBatch( adapter, { runId: "run-1", organizationId: "org-1", projectId: "proj-1", workspaceDir: "/tmp/ws", prisma: {} as never, }, [ { inputPath: "a.pdf", outputDir: "out/same" }, { inputPath: "b.pdf", outputDir: "out/same/" }, ], 2, )).rejects.toThrow(/distinct output_dir/); expect(invoke).not.toHaveBeenCalled(); }); });