curriculum-project-hub/hub/test/unit/pdf-to-md-batch.test.ts

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();
  });
});