From db49a0d23dfe46a77bd04f77db5ad40c78ac151e Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Tue, 21 Jul 2026 05:55:55 +0000 Subject: [PATCH] feat(hub): concurrent multi-PDF convert_pdf_to_md + readable skills (v0.0.40) Teachers convert many PDFs in one tool call with bounded Docmind concurrency. Each item keeps its own output_dir/document.md and UsageFact; failures are per-file. Mirror role skills to .cph/runtime-skills and CPH_RUNTIME_SKILLS_DIR so agents can Read SKILL.md instead of dead .claude/sandbox stubs. --- hub/.env.example | 2 + hub/deploy/install_service.sh | 1 + hub/package-lock.json | 4 +- hub/package.json | 2 +- hub/skills/pdf-to-md/SKILL.md | 71 ++++++++---- hub/src/agent/security.ts | 21 +++- hub/src/capability/pdfToMdBundle.ts | 129 ++++++++++++++++++++++ hub/src/feishu/fileDeliveryTool.ts | 93 +++++++++++++--- hub/test/unit/agent-security.test.ts | 40 ++++++- hub/test/unit/pdf-to-md-batch.test.ts | 148 ++++++++++++++++++++++++++ 10 files changed, 475 insertions(+), 36 deletions(-) create mode 100644 hub/test/unit/pdf-to-md-batch.test.ts diff --git a/hub/.env.example b/hub/.env.example index c074f57..b528fd2 100644 --- a/hub/.env.example +++ b/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" diff --git a/hub/deploy/install_service.sh b/hub/deploy/install_service.sh index 53d321d..70c32ef 100755 --- a/hub/deploy/install_service.sh +++ b/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 diff --git a/hub/package-lock.json b/hub/package-lock.json index a065c6e..225dfb7 100644 --- a/hub/package-lock.json +++ b/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", diff --git a/hub/package.json b/hub/package.json index 0f77428..f52ffb0 100644 --- a/hub/package.json +++ b/hub/package.json @@ -1,6 +1,6 @@ { "name": "@paradigm/hub", - "version": "0.0.39", + "version": "0.0.40", "private": true, "type": "module", "engines": { diff --git a/hub/skills/pdf-to-md/SKILL.md b/hub/skills/pdf-to-md/SKILL.md index af0a6b9..f6a53a7 100644 --- a/hub/skills/pdf-to-md/SKILL.md +++ b/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. diff --git a/hub/src/agent/security.ts b/hub/src/agent/security.ts index 62e326e..44b9ffe 100644 --- a/hub/src/agent/security.ts +++ b/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, diff --git a/hub/src/capability/pdfToMdBundle.ts b/hub/src/capability/pdfToMdBundle.ts index 2b096ff..5b03395 100644 --- a/hub/src/capability/pdfToMdBundle.ts +++ b/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> = 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( + items: readonly T[], + concurrency: number, + worker: (item: T, index: number) => Promise, +): Promise { + if (items.length === 0) return []; + const limit = Math.max(1, Math.min(Math.trunc(concurrency), items.length)); + const results = new Array(items.length); + let next = 0; + async function runWorker(): Promise { + 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, + items: readonly PdfToMdBatchItem[], + concurrency: number = DEFAULT_PDF_TO_MD_CONCURRENCY, +): Promise { + 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(); + 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 { diff --git a/hub/src/feishu/fileDeliveryTool.ts b/hub/src/feishu/fileDeliveryTool.ts index 58a86c6..b2eaa32 100644 --- a/hub/src/feishu/fileDeliveryTool.ts +++ b/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): string { const instructions: string[] = []; if (enabledTools.has("send_file")) { @@ -315,10 +380,14 @@ function mcpInstructions(enabledTools: ReadonlySet): 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.md or $CPH_RUNTIME_SKILLS_DIR//SKILL.md. Prefer the Skill tool when available. Workspace .claude/ and .mcp.json are sandbox stubs — not skill or MCP source.", + ); return instructions.join(" "); } diff --git a/hub/test/unit/agent-security.test.ts b/hub/test/unit/agent-security.test.ts index 7edd53e..43be8fb 100644 --- a/hub/test/unit/agent-security.test.ts +++ b/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"); diff --git a/hub/test/unit/pdf-to-md-batch.test.ts b/hub/test/unit/pdf-to-md-batch.test.ts new file mode 100644 index 0000000..1ec0b7f --- /dev/null +++ b/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(); + }); +});