feat(hub): concurrent multi-PDF convert_pdf_to_md + readable skills (v0.0.40) (#20)

hub/.env.example

@@ -26,6 +26,8 @@ HUB_AGENT_MAX_RUN_SECONDS="1800"
 HUB_HTTP_BODY_LIMIT_BYTES="1048576"
 HUB_MAX_FILES_PER_MESSAGE="20"
 HUB_MAX_FILE_BYTES="26214400"
+# Max concurrent Alibaba Docmind jobs for one convert_pdf_to_md batch (1-8).
+HUB_PDF_TO_MD_MAX_CONCURRENT="3"
 HUB_HTTP_REQUESTS_PER_MINUTE="120"
 HUB_FEISHU_EVENTS_PER_MINUTE="120"
 

hub/deploy/install_service.sh

@@ -175,6 +175,7 @@ HUB_AGENT_MAX_RUN_SECONDS=
 HUB_HTTP_BODY_LIMIT_BYTES=
 HUB_MAX_FILES_PER_MESSAGE=
 HUB_MAX_FILE_BYTES=
+HUB_PDF_TO_MD_MAX_CONCURRENT=3
 HUB_HTTP_REQUESTS_PER_MINUTE=
 HUB_FEISHU_EVENTS_PER_MINUTE=
 HUB_FEISHU_LISTENER_ENABLED=true

hub/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@paradigm/hub",
-  "version": "0.0.39",
+  "version": "0.0.40",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@paradigm/hub",
-      "version": "0.0.39",
+      "version": "0.0.40",
       "dependencies": {
         "@alicloud/credentials": "^2.4.5",
         "@alicloud/docmind-api20220711": "^1.4.15",

hub/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@paradigm/hub",
-  "version": "0.0.39",
+  "version": "0.0.40",
   "private": true,
   "type": "module",
   "engines": {

hub/skills/pdf-to-md/SKILL.md

@@ -2,62 +2,95 @@
 name: pdf-to-md
 description: >
   Convert PDF documents to Markdown bundles using the convert_pdf_to_md tool.
-  Handles PDFs from Feishu messages, local workspace files, and produces
-  high-quality Markdown with LaTeX formulas and extracted images.
+  Handles single or multiple PDFs (concurrent batch), Feishu attachments, and
+  local workspace files. Produces high-quality Markdown with LaTeX formulas
+  and extracted images.
 ---
 
 # PDF to Markdown Conversion
 
 ## When to use
 
-Use this skill when the user asks to convert a PDF to Markdown, extract text
-from a PDF, or turn a PDF document into an editable format.
+Use this skill when the user asks to convert a PDF (or several PDFs) to
+Markdown, extract text from a PDF, or turn PDF documents into an editable
+format.
 
 ## How it works
 
-The `convert_pdf_to_md` tool (provided by the `cph_hub` MCP server) calls
-Alibaba Cloud Document Mind to parse the PDF. It:
+The `convert_pdf_to_md` tool (provided by the in-process `cph_hub` MCP server)
+calls Alibaba Cloud Document Mind to parse each PDF. It:
 
 - Extracts text in reading order (handles multi-column, scanned, and
   multi-language documents)
 - Converts mathematical formulas to **LaTeX** (`$...$` inline, `$$...$$` block)
 - Extracts tables as Markdown tables
 - Downloads embedded images into the output directory
-- Writes a single `document.md` file plus image files
+- Writes a single `document.md` file plus image files **per** `output_dir`
+
+There is **no** workspace `.mcp.json` source file. MCP tools are injected by
+Hub at run start. Do not look for MCP or skill source under workspace
+`.claude/` — those paths are sandbox stubs (often character devices) and are
+not readable constitution.
+
+## Where this skill text lives
+
+Prefer the Skill tool when the runtime offers it. If you need to re-read these
+instructions with Read:
+
+- Workspace copy (always under cwd): `.cph/runtime-skills/pdf-to-md/SKILL.md`
+- Absolute path env: `$CPH_RUNTIME_SKILLS_DIR/pdf-to-md/SKILL.md`
 
 ## Workflow
 
-### PDF from a Feishu message
+### One PDF from a Feishu message
 
 1. Use `feishu_read_context` to find the `file_key` of the PDF attachment.
 2. Use `feishu_download_resource` to download it into the workspace.
-3. Use `convert_pdf_to_md` with the downloaded file path and an output directory.
+3. Use `convert_pdf_to_md` with `input_path` + `output_dir`.
 
 ### PDF already in the workspace
 
-1. Use `convert_pdf_to_md` directly with the file path and an output directory.
+1. Use `convert_pdf_to_md` with `input_path` and `output_dir`.
+
+### Multiple PDFs (concurrent)
+
+1. Download or locate every PDF in the workspace first.
+2. Call **`convert_pdf_to_md` once** with:
+
+```json
+{
+  "items": [
+    { "input_path": "sources/a.pdf", "output_dir": "md/a" },
+    { "input_path": "sources/b.pdf", "output_dir": "md/b" }
+  ]
+}
+```
+
+3. Hub submits Docmind jobs with bounded concurrency (default 3, max 8;
+   optional `concurrency` argument). Prefer this over N sequential tool calls.
+4. **Each item must use a distinct `output_dir`** — the tool always writes
+   `document.md` inside that directory; shared dirs overwrite each other.
+5. Partial failure returns per-file OK/FAIL lines; re-run only failed items.
 
 ## Important rules
 
 - **Always** use `convert_pdf_to_md` for PDF→Markdown. Do NOT attempt to parse
-  PDFs yourself with Read, Bash, Python, or any other method. The tool provides
-  accurate formula, table, and image extraction that manual methods cannot
-  match.
+  PDFs yourself with Read, Bash, Python, or any other method.
 - If `convert_pdf_to_md` fails because no capability connection is configured,
   tell the user to ask their organization admin to configure the Aliyun
   docmind credential in the admin web UI (组织后台 → 能力).
 - The output directory will be created if it does not exist.
-- After conversion, use `send_file` to send the generated markdown back to the
-  user if they requested it.
+- After conversion, use `send_file` to send generated markdown (or a zip you
+  assemble) back to the user if they requested delivery.
 
 ## Output
 
-The tool returns a list of generated files:
+Per `output_dir`:
+
 - `document.md` — the main markdown file
 - `*.jpg` / `*.png` — extracted images, referenced from the markdown
 
 ## Cost
 
-The conversion is billed per page (0.04 CNY/page ≈ $0.0056/page for the
-enhanced formula mode). The cost is automatically recorded on the run's
-usage ledger.
+Billed per page (0.04 CNY/page ≈ $0.0056/page for enhanced formula mode).
+Each successful file records its own usage fact on the run ledger.

hub/src/agent/security.ts

@@ -1,4 +1,4 @@
-import { chmod, lstat, mkdir, realpath } from "node:fs/promises";
+import { chmod, cp, lstat, mkdir, readdir, realpath, rm } from "node:fs/promises";
 import { homedir } from "node:os";
 import { isAbsolute, join, relative, resolve } from "node:path";
 import type { RoleSkillEntry } from "./models.js";
@@ -142,6 +142,25 @@ export async function createAgentSecurityPolicy(input: AgentSecurityInput): Prom
         runId: input.runId,
         skills: selectedSkills,
       });
+  // Mirror selected skills under the workspace so the agent can Read SKILL.md
+  // without guessing the opaque host plugin UUID path. Workspace `.claude/` and
+  // `.mcp.json` are Claude sandbox stubs — not skill/MCP source of truth.
+  const runtimeSkillsRel = join(".cph", "runtime-skills");
+  const runtimeSkillsAbs = join(workspaceDir, runtimeSkillsRel);
+  await rm(runtimeSkillsAbs, { recursive: true, force: true });
+  await mkdir(runtimeSkillsAbs, { recursive: true, mode: 0o700 });
+  if (skillPlugin !== null) {
+    const pluginSkillsRoot = join(skillPlugin.root, "skills");
+    const skillNames = await readdir(pluginSkillsRoot);
+    for (const skillName of skillNames) {
+      await cp(join(pluginSkillsRoot, skillName), join(runtimeSkillsAbs, skillName), {
+        recursive: true,
+        force: true,
+      });
+    }
+    env.CPH_RUNTIME_SKILLS_DIR = runtimeSkillsAbs;
+    env.CPH_RUNTIME_SKILLS_REL = runtimeSkillsRel;
+  }
   return {
     cwd: workspaceDir,
     workspaceRoot,

hub/src/capability/pdfToMdBundle.ts

@@ -63,6 +63,135 @@ export interface PdfToMdBundleDeps {
   readonly prisma: PrismaClient;
 }
 
+/** Default max concurrent Docmind jobs for one convert_pdf_to_md batch call. */
+export const DEFAULT_PDF_TO_MD_CONCURRENCY = 3;
+/** Hard ceiling for agent-requested concurrency (also clamps env). */
+export const MAX_PDF_TO_MD_CONCURRENCY = 8;
+/** Max PDFs accepted in one batch tool call. */
+export const MAX_PDF_TO_MD_BATCH_ITEMS = 32;
+
+export function clampPdfToMdConcurrency(value: number): number {
+  if (!Number.isFinite(value)) return DEFAULT_PDF_TO_MD_CONCURRENCY;
+  const n = Math.trunc(value);
+  if (n < 1) return 1;
+  if (n > MAX_PDF_TO_MD_CONCURRENCY) return MAX_PDF_TO_MD_CONCURRENCY;
+  return n;
+}
+
+/** Read HUB_PDF_TO_MD_MAX_CONCURRENT (default 3, max 8). */
+export function readPdfToMdConcurrency(
+  env: Readonly<Record<string, string | undefined>> = process.env,
+): number {
+  const raw = env["HUB_PDF_TO_MD_MAX_CONCURRENT"]?.trim();
+  if (raw === undefined || raw === "") return DEFAULT_PDF_TO_MD_CONCURRENCY;
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed < 1) {
+    throw new Error(`HUB_PDF_TO_MD_MAX_CONCURRENT must be a positive integer, got ${raw}`);
+  }
+  return clampPdfToMdConcurrency(parsed);
+}
+
+export interface PdfToMdBatchItem {
+  readonly inputPath: string;
+  readonly outputDir: string;
+}
+
+export type PdfToMdBatchItemResult =
+  | {
+      readonly ok: true;
+      readonly inputPath: string;
+      readonly outputDir: string;
+      readonly result: CapabilityInvocationResult;
+    }
+  | {
+      readonly ok: false;
+      readonly inputPath: string;
+      readonly outputDir: string;
+      readonly error: string;
+    };
+
+/**
+ * Run worker over items with bounded parallelism. Order of results matches
+ * input order. Rejects in worker are not swallowed — caller should catch.
+ */
+export async function mapPool<T, R>(
+  items: readonly T[],
+  concurrency: number,
+  worker: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  if (items.length === 0) return [];
+  const limit = Math.max(1, Math.min(Math.trunc(concurrency), items.length));
+  const results = new Array<R>(items.length);
+  let next = 0;
+  async function runWorker(): Promise<void> {
+    for (;;) {
+      const index = next;
+      next += 1;
+      if (index >= items.length) return;
+      results[index] = await worker(items[index]!, index);
+    }
+  }
+  await Promise.all(Array.from({ length: limit }, () => runWorker()));
+  return results;
+}
+
+/**
+ * Convert multiple PDFs with bounded concurrency. Each item is attribute-
+ * independent (own paths + own UsageFact). Failures are per-item and do not
+ * cancel siblings; order matches `items`.
+ */
+export async function invokePdfToMdBatch(
+  adapter: CapabilityAdapter,
+  base: Omit<CapabilityInvocationInput, "inputPath" | "outputDir">,
+  items: readonly PdfToMdBatchItem[],
+  concurrency: number = DEFAULT_PDF_TO_MD_CONCURRENCY,
+): Promise<PdfToMdBatchItemResult[]> {
+  if (items.length === 0) {
+    throw new Error("pdf_to_md batch requires at least one item");
+  }
+  if (items.length > MAX_PDF_TO_MD_BATCH_ITEMS) {
+    throw new Error(
+      `pdf_to_md batch supports at most ${MAX_PDF_TO_MD_BATCH_ITEMS} items per call (got ${items.length})`,
+    );
+  }
+  const seenOutputDirs = new Set<string>();
+  for (const item of items) {
+    if (item.inputPath.trim() === "" || item.outputDir.trim() === "") {
+      throw new Error("pdf_to_md batch items require non-empty inputPath and outputDir");
+    }
+    const key = item.outputDir.replace(/\\/g, "/").replace(/\/+$/, "");
+    if (seenOutputDirs.has(key)) {
+      throw new Error(
+        `pdf_to_md batch items must use distinct output_dir values; duplicate: ${item.outputDir}`,
+      );
+    }
+    seenOutputDirs.add(key);
+  }
+  const limit = clampPdfToMdConcurrency(concurrency);
+  return mapPool(items, limit, async (item) => {
+    try {
+      const result = await adapter.invoke({
+        ...base,
+        inputPath: item.inputPath,
+        outputDir: item.outputDir,
+      });
+      return {
+        ok: true as const,
+        inputPath: item.inputPath,
+        outputDir: item.outputDir,
+        result,
+      };
+    } catch (error) {
+      return {
+        ok: false as const,
+        inputPath: item.inputPath,
+        outputDir: item.outputDir,
+        error: error instanceof Error ? error.message : String(error),
+      };
+    }
+  });
+}
+
 /** Build the pdf_to_md_bundle adapter. The client is injectable for testing. */
 export function createPdfToMdBundleAdapter(deps: PdfToMdBundleDeps): CapabilityAdapter {
   return {

hub/src/feishu/fileDeliveryTool.ts

@@ -9,7 +9,13 @@ import { CPH_HUB_MCP_TOOL_IDS, type CphHubMcpToolId } from "../agent/roleTools.j
 import { WorkspaceFileBoundaryError } from "../security/workspaceFiles.js";
 import type { PrismaClient } from "@prisma/client";
 import type { LocalSecretEnvelope } from "../security/secretEnvelope.js";
-import { createPdfToMdBundleAdapter } from "../capability/pdfToMdBundle.js";
+import {
+  createPdfToMdBundleAdapter,
+  invokePdfToMdBatch,
+  readPdfToMdConcurrency,
+  MAX_PDF_TO_MD_BATCH_ITEMS,
+  type PdfToMdBatchItemResult,
+} from "../capability/pdfToMdBundle.js";
 import { AliyunDocmindClient } from "../capability/docmindClient.js";
 
 export interface FileDeliveryToolOptions {
@@ -246,21 +252,54 @@ export function createFileDeliveryMcpServer(options: FileDeliveryToolOptions): M
     tools.push(
       tool(
         "convert_pdf_to_md",
-        "Convert a PDF file in the workspace to a Markdown bundle (markdown + extracted images) using Alibaba Cloud Document Mind. The PDF must already be in the workspace (use feishu_download_resource first if it came from Feishu). Returns the path to the generated markdown file and the list of extracted image paths. Mathematical formulas are converted to LaTeX.",
+        "Convert one or more PDF files in the workspace to Markdown bundles (markdown + extracted images) via Alibaba Cloud Document Mind. PDFs must already be in the workspace (use feishu_download_resource first for Feishu attachments). Prefer a single call with `items` for multiple PDFs — Hub converts them concurrently (bounded). Each item needs its own output_dir because the tool always writes document.md inside that directory. Formulas become LaTeX.",
         {
-          input_path: z.string().describe("Relative path to the input PDF within the workspace."),
-          output_dir: z.string().describe("Relative directory within the workspace to write the markdown and images into. Will be created if it does not exist."),
+          input_path: z.string().optional().describe("Single-file mode: workspace-relative path to the input PDF. Required when `items` is omitted."),
+          output_dir: z.string().optional().describe("Single-file mode: workspace-relative directory for document.md + images. Required when `items` is omitted."),
+          items: z.array(z.object({
+            input_path: z.string().describe("Workspace-relative path to one input PDF."),
+            output_dir: z.string().describe("Workspace-relative output directory for this PDF (must be unique per item)."),
+          })).min(1).max(MAX_PDF_TO_MD_BATCH_ITEMS).optional().describe(`Batch mode: multiple PDFs converted concurrently. Max ${MAX_PDF_TO_MD_BATCH_ITEMS} items. Do not reuse output_dir across items.`),
+          concurrency: z.number().int().min(1).max(8).optional().describe("Optional parallel job limit for batch mode (1-8). Defaults to HUB_PDF_TO_MD_MAX_CONCURRENT (usually 3)."),
         },
         async (args) => {
+          const base = {
+            runId: options.runId,
+            organizationId: options.organizationId,
+            projectId: options.projectId,
+            workspaceDir: options.workspaceDir,
+            prisma: options.prisma,
+          };
           try {
+            if (args.items !== undefined && args.items.length > 0) {
+              const batchResults = await invokePdfToMdBatch(
+                adapter,
+                base,
+                args.items.map((item) => ({
+                  inputPath: item.input_path,
+                  outputDir: item.output_dir,
+                })),
+                args.concurrency ?? readPdfToMdConcurrency(),
+              );
+              return {
+                content: [{ type: "text", text: formatPdfToMdBatchResult(batchResults) }],
+                ...(batchResults.every((item) => item.ok) ? {} : { isError: true }),
+              };
+            }
+            if (args.input_path === undefined || args.input_path.trim() === ""
+              || args.output_dir === undefined || args.output_dir.trim() === "") {
+              return {
+                isError: true,
+                content: [{
+                  type: "text",
+                  text: "convert_pdf_to_md requires either items[{input_path,output_dir},...] or both input_path and output_dir.",
+                }],
+              };
+            }
             const result = await adapter.invoke({
-              runId: options.runId,
-              organizationId: options.organizationId,
-              projectId: options.projectId,
-              workspaceDir: options.workspaceDir,
+              ...base,
               inputPath: args.input_path,
               outputDir: args.output_dir,
-              prisma: options.prisma,
             });
             const lines = [`Converted PDF to markdown. ${result.artifacts.length} files written:`];
             for (const artifact of result.artifacts) {
@@ -292,6 +331,32 @@ export function createFileDeliveryMcpServer(options: FileDeliveryToolOptions): M
   });
 }
 
+
+function formatPdfToMdBatchResult(results: readonly PdfToMdBatchItemResult[]): string {
+  const ok = results.filter((item) => item.ok);
+  const failed = results.filter((item) => !item.ok);
+  const lines = [
+    `Batch PDF→Markdown finished: ${ok.length} succeeded, ${failed.length} failed (of ${results.length}).`,
+  ];
+  for (const item of results) {
+    if (item.ok) {
+      const md = item.result.artifacts.find((artifact) => artifact.kind === "markdown")?.path;
+      lines.push(
+        `OK ${item.inputPath} → ${item.outputDir}`
+          + (md !== undefined ? ` (${md})` : "")
+          + `; pages=${item.result.consumption.quantity}`
+          + `; cost=$${(item.result.consumption.costUsd ?? 0).toFixed(4)}`,
+      );
+      for (const artifact of item.result.artifacts) {
+        lines.push(`  - ${artifact.path} (${artifact.kind})`);
+      }
+    } else {
+      lines.push(`FAIL ${item.inputPath} → ${item.outputDir}: ${item.error}`);
+    }
+  }
+  return lines.join("\n");
+}
+
 function mcpInstructions(enabledTools: ReadonlySet<CphHubMcpToolId>): string {
   const instructions: string[] = [];
   if (enabledTools.has("send_file")) {
@@ -315,10 +380,14 @@ function mcpInstructions(enabledTools: ReadonlySet<CphHubMcpToolId>): string {
   }
   if (enabledTools.has("convert_pdf_to_md")) {
     instructions.push(
-      "Use convert_pdf_to_md when the user asks to convert a PDF to Markdown.",
-      "If the PDF came from a Feishu message, first use feishu_download_resource to save it to the workspace, then call convert_pdf_to_md.",
-      "Do NOT attempt to parse PDFs yourself with Read or Bash — always use convert_pdf_to_md for accurate text, formula, and image extraction.",
+      "Use convert_pdf_to_md when the user asks to convert a PDF (or several PDFs) to Markdown.",
+      "If PDFs came from Feishu, download each with feishu_download_resource first, then convert.",
+      "For multiple PDFs, call convert_pdf_to_md once with items=[{input_path,output_dir},...] so Hub converts them concurrently; give each file its own output_dir (the tool writes document.md inside it).",
+      "Do NOT attempt to parse PDFs yourself with Read or Bash — always use convert_pdf_to_md.",
     );
   }
+  instructions.push(
+    "Role skill docs (when bound) are readable at .cph/runtime-skills/<skill-name>/SKILL.md or $CPH_RUNTIME_SKILLS_DIR/<skill-name>/SKILL.md. Prefer the Skill tool when available. Workspace .claude/ and .mcp.json are sandbox stubs — not skill or MCP source.",
+  );
   return instructions.join(" ");
 }

hub/test/unit/agent-security.test.ts

@@ -1,8 +1,9 @@
-import { mkdir, mkdtemp, realpath, rm, symlink } from "node:fs/promises";
+import { mkdir, mkdtemp, readFile, realpath, rm, symlink, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, describe, expect, it } from "vitest";
 import { createAgentSecurityPolicy } from "../../src/agent/security.js";
+import { importSkillDirectory } from "../../src/agent/skillStore.js";
 
 describe("agent subprocess security policy", () => {
   const roots: string[] = [];
@@ -135,6 +136,43 @@ describe("agent subprocess security policy", () => {
     })).rejects.toThrow("Agent temp path is too long for sandbox bridge sockets");
   });
 
+
+  it("mirrors selected skills under .cph/runtime-skills and exposes CPH_RUNTIME_SKILLS_DIR", async () => {
+    const { root, workspaceRoot, workspace } = await makeWorkspace();
+    const storeRoot = join(root, "skills-store");
+    const skillSource = join(root, "skill-src", "pdf-to-md");
+    await mkdir(skillSource, { recursive: true });
+    await writeFile(
+      join(skillSource, "SKILL.md"),
+      "---\nname: pdf-to-md\ndescription: convert\n---\n# pdf-to-md\nbatch items\n",
+    );
+    const installed = await importSkillDirectory({ sourceDir: skillSource, storeRoot });
+    const policy = await createAgentSecurityPolicy({
+      runId: "run-skills",
+      workspaceRoot,
+      workspaceDir: workspace,
+      skills: [{
+        name: "pdf-to-md",
+        version: "1",
+        contentDigest: installed.contentDigest,
+      }],
+      hostEnv: {
+        PATH: "/usr/bin:/bin",
+        HUB_SKILL_STORE_ROOT: storeRoot,
+      },
+    });
+    const canonicalWorkspace = await realpath(workspace);
+    const mirrored = join(canonicalWorkspace, ".cph", "runtime-skills", "pdf-to-md", "SKILL.md");
+    await expect(readFile(mirrored, "utf8")).resolves.toContain("batch items");
+    expect(policy.env.CPH_RUNTIME_SKILLS_DIR).toBe(join(canonicalWorkspace, ".cph", "runtime-skills"));
+    expect(policy.env.CPH_RUNTIME_SKILLS_REL).toBe(join(".cph", "runtime-skills"));
+    expect(policy.skillIds).toEqual(["cph-runtime:pdf-to-md"]);
+    expect(policy.sandbox.filesystem.allowRead).toEqual(
+      expect.arrayContaining([canonicalWorkspace, policy.skillPluginRoot]),
+    );
+    await policy.cleanup();
+  });
+
   it("rejects a project workspace whose real path escapes the configured workspace root", async () => {
     const { root, workspaceRoot } = await makeWorkspace();
     const outside = join(root, "outside");

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

@@ -0,0 +1,148 @@
+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();
+  });
+});