curriculum-project-hub/hub/src/capability/pdfToMdBundle.ts

/**
 * ADR-0027: pdf_to_md_bundle capability adapter.
 *
 * Converts a PDF in the run's workspace into a Markdown bundle (md + extracted
 * images) by calling the MinerU document parsing service, writing outputs into
 * the workspace, and recording consumption on a UsageFact (ADR-0026).
 *
 * Invariants (ADR-0027):
 *   1. Credential isolation — the capability credential is resolved in Hub and
 *      never reaches the Agent process. The MineruClient receives it as a
 *      call argument, not from the environment.
 *   2. Workspace containment — input and output paths are confined to the
 *      run's workspace dir (ADR-0018 AgentSurface). Escapes are rejected.
 *   3. Mandatory fact — a successful invocation always writes ≥1 UsageFact
 *      with kind=external_capability, even when costUsd is null (ADR-0022:
 *      missing cost ≠ zero).
 */
import { mkdir, writeFile } from "node:fs/promises";
import { join, resolve, relative, isAbsolute } from "node:path";
import type { PrismaClient } from "@prisma/client";
import { LocalSecretEnvelope } from "../security/secretEnvelope.js";
import { resolveCapabilityCredential } from "./capabilityConnections.js";
import { DocmindClientError, type CapabilityProviderClient } from "./docmindClient.js";
import {
  CAPABILITIES,
  asDocmindSecret,
  type CapabilityAdapter,
  type CapabilityInvocationInput,
  type CapabilityInvocationResult,
  type CapabilityOutputArtifact,
} from "./types.js";

const CAPABILITY_ID = "pdf_to_md_bundle" as const;
const PROVIDER_ID = "aliyun_docmind";

/** Thrown when a requested path escapes the workspace root (ADR-0018). */
export class CapabilityPathEscape extends Error {
  constructor(readonly requested: string, readonly workspaceDir: string) {
    super(`capability path escapes workspace: ${requested} (root ${workspaceDir})`);
    this.name = "CapabilityPathEscape";
  }
}

/** Resolve a workspace-relative path, rejecting escapes (ADR-0018 AgentSurface). */
function confineToWorkspace(requestedPath: string, workspaceDir: string): string {
  if (isAbsolute(requestedPath)) {
    const rel = relative(workspaceDir, requestedPath);
    if (rel.startsWith("..") || rel === "") {
      throw new CapabilityPathEscape(requestedPath, workspaceDir);
    }
    return requestedPath;
  }
  const resolved = resolve(workspaceDir, requestedPath);
  const rel = relative(workspaceDir, resolved);
  if (rel.startsWith("..")) {
    throw new CapabilityPathEscape(requestedPath, workspaceDir);
  }
  return resolved;
}

export interface PdfToMdBundleDeps {
  readonly secrets: LocalSecretEnvelope;
  readonly client: CapabilityProviderClient;
  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 {
    capabilityId: CAPABILITY_ID,
    async invoke(input: CapabilityInvocationInput): Promise<CapabilityInvocationResult> {
      const descriptor = CAPABILITIES[CAPABILITY_ID];
      // 1. Resolve org-scoped credential (fail-closed, ADR-0024/0027).
      const credential = await resolveCapabilityCredential(deps.prisma, deps.secrets, {
        organizationId: input.organizationId,
        capabilityId: CAPABILITY_ID,
      });

      // 2. Confine input + output paths to the workspace (ADR-0018).
      const absoluteInput = confineToWorkspace(input.inputPath, input.workspaceDir);
      const absoluteOutputDir = confineToWorkspace(input.outputDir, input.workspaceDir);
      await mkdir(absoluteOutputDir, { recursive: true });

      // 3. Call the backing service.
      let result;
      try {
        result = await deps.client.parse(asDocmindSecret(credential), { inputFilePath: absoluteInput });
      } catch (e) {
        if (e instanceof DocmindClientError) throw e;
        throw new DocmindClientError(
          e instanceof Error ? e.message : String(e),
          "docmind_unreachable",
        );
      }
      if (result.markdown === "") {
        throw new DocmindClientError("docmind returned empty markdown", "docmind_no_output");
      }

      // 4. Write outputs into the workspace.
      const artifacts: CapabilityOutputArtifact[] = [];
      const mdPath = join(absoluteOutputDir, "document.md");
      await writeFile(mdPath, result.markdown, "utf8");
      artifacts.push({ path: relative(input.workspaceDir, mdPath), kind: "markdown" });

      for (const image of result.images) {
        const imagePath = join(absoluteOutputDir, image.filename);
        const rel = relative(absoluteOutputDir, imagePath);
        if (rel.startsWith("..")) {
          // Defensive: image filename must not escape the output dir.
          throw new CapabilityPathEscape(image.filename, absoluteOutputDir);
        }
        await writeFile(imagePath, image.data);
        artifacts.push({ path: relative(input.workspaceDir, imagePath), kind: "image" });
      }

      // 5. Write the UsageFact (ADR-0026/0027). Always written on success;
      //    costUsd null means unknown, NOT zero (ADR-0022).
      const occurredAt = new Date();
      await deps.prisma.usageFact.create({
        data: {
          runId: input.runId,
          occurredAt,
          kind: "external_capability",
          provider: PROVIDER_ID,
          model: null,
          inputTokens: null,
          outputTokens: null,
          quantity: result.pageCount,
          unit: descriptor.meteringUnit,
          costUsd: result.costUsd,
          costSource: result.costUsd !== null ? "provider_reported" : "unknown",
          capabilityId: CAPABILITY_ID,
          correlationId: result.requestId,
          metadata: {},
        },
      });

      return {
        artifacts,
        consumption: {
          provider: PROVIDER_ID,
          model: null,
          inputTokens: null,
          outputTokens: null,
          quantity: result.pageCount,
          unit: descriptor.meteringUnit,
          costUsd: result.costUsd,
          correlationId: result.requestId,
        },
      };
    },
  };
}