fix: validate non-image inputs by modality (batch + pipeline) (#244)

* fix(batch): validate non-image inputs by modality

Batch processing ran validateImageBuffer and the image-only decode chain
on every uploaded file, so audio, video, and document tools rejected all
inputs with "Invalid image: Unrecognized image format" and returned
422 "All files failed processing".

Resolve the tool's modality and route non-image files through their own
input handler (inputHandlerFor(modality).prepare), mirroring the
single-file path. The image batch path is unchanged.

* fix(pipeline): validate non-image inputs by modality

The pipeline /execute and /batch routes validated every upload with
validateImageBuffer and ran the image-only decode chain, so audio,
video, and document pipelines were rejected with "Invalid image:
Unrecognized image format".

Resolve the input modality from the first step's tool and route
non-image inputs through their modality handler, mirroring the batch
and single-file paths. Image pipelines are unchanged.

apps/api/src/routes/batch.ts

@@ -9,7 +9,10 @@
  * Returns a ZIP file containing all processed images.
  */
 import { randomUUID } from "node:crypto";
-import { getBundleForTool, TOOL_BUNDLE_MAP } from "@snapotter/shared";
+import { mkdir } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { getBundleForTool, TOOL_BUNDLE_MAP, TOOLS } from "@snapotter/shared";
 import archiver from "archiver";
 import type { FlowJob } from "bullmq";
 import { eq } from "drizzle-orm";
@@ -30,6 +33,8 @@ import { decodeToSharpCompat, needsCliDecode } from "../lib/format-decoders.js";
 import { decodeHeic } from "../lib/heic-converter.js";
 import { getObjectStream, putObject } from "../lib/object-storage.js";
 import { resolveToolPool } from "../lib/pool.js";
+import { InputValidationError } from "../modality/contract.js";
+import { inputHandlerFor } from "../modality/input-handler.js";
 import { getAuthUser } from "../plugins/auth.js";
 import { updateJobProgress } from "./progress.js";
 import { getToolConfig } from "./tool-factory.js";
@@ -175,11 +180,18 @@ export async function registerBatchRoutes(app: FastifyInstance): Promise<void> {
       const preFailures: Array<{ originalIndex: number; filename: string; error: string }> = [];
       let flowChildIndex = 0;
 
+      // Resolve the tool's modality so non-image files (audio/video/document)
+      // validate through their own handler instead of the image validator.
+      const modality = TOOLS.find((t) => t.id === toolId)?.modality ?? "image";
+      const batchScratch = join(tmpdir(), "snapotter-scratch", `batch-${parentId}`);
+      await mkdir(batchScratch, { recursive: true });
+
       for (let i = 0; i < files.length; i++) {
         const file = files[i];
         let processBuffer = file.buffer;
         let processFilename = file.filename;
 
+        if (modality === "image") {
           const validation = await validateImageBuffer(processBuffer, processFilename);
           if (!validation.valid) {
             preFailures.push({
@@ -226,6 +238,29 @@ export async function registerBatchRoutes(app: FastifyInstance): Promise<void> {
           if (!skipPreprocess) {
             processBuffer = await autoOrient(processBuffer);
           }
+        } else {
+          // Non-image modalities (audio/video/document/file): validate and
+          // decode through the tool's own input handler, mirroring the
+          // single-file path. Previously batch validated everything as an
+          // image, which rejected all audio/video/document inputs.
+          try {
+            const prepared = await inputHandlerFor(modality).prepare(
+              processBuffer,
+              processFilename,
+              {
+                scratchDir: batchScratch,
+              },
+            );
+            processBuffer = prepared.buffer;
+            processFilename = prepared.filename;
+          } catch (err) {
+            if (err instanceof InputValidationError) {
+              preFailures.push({ originalIndex: i, filename: file.filename, error: err.message });
+              continue;
+            }
+            throw err;
+          }
+        }
 
         // Upload decoded file to object storage
         const childId = `${parentId}-f${flowChildIndex}`;

apps/api/src/routes/pipeline.ts

@@ -8,7 +8,10 @@
  * POST   /api/v1/pipeline/batch    -- Batch pipeline execution (ZIP output)
  */
 import { randomUUID } from "node:crypto";
-import { ANALYTICS_EVENTS, getBundleForTool, TOOL_BUNDLE_MAP } from "@snapotter/shared";
+import { mkdir } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { ANALYTICS_EVENTS, getBundleForTool, TOOL_BUNDLE_MAP, TOOLS } from "@snapotter/shared";
 import archiver from "archiver";
 import type { FlowJob } from "bullmq";
 import { eq } from "drizzle-orm";
@@ -31,6 +34,8 @@ import { decodeHeic } from "../lib/heic-converter.js";
 import { getObjectStream, putObject } from "../lib/object-storage.js";
 import { resolveToolPool } from "../lib/pool.js";
 import { isSvgBuffer, sanitizeSvg } from "../lib/svg-sanitize.js";
+import { InputValidationError } from "../modality/contract.js";
+import { inputHandlerFor } from "../modality/input-handler.js";
 import { hasEffectivePermission } from "../permissions.js";
 import { getAuthUser, requireAuth } from "../plugins/auth.js";
 import { updateJobProgress, updateSingleFileProgress } from "./progress.js";
@@ -232,9 +237,24 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
     }
 
     if (!fileBuffer || fileBuffer.length === 0) {
-      return reply.status(400).send({ error: "No image file provided" });
+      return reply.status(400).send({ error: "No file provided" });
     }
 
+    // The first pipeline step determines the input modality, so non-image
+    // inputs (audio/video/document) get validated by the right handler instead
+    // of always being forced through image validation/decoding.
+    let firstToolId: string | undefined;
+    try {
+      firstToolId = (JSON.parse(pipelineRaw ?? "{}") as { steps?: Array<{ toolId?: string }> })
+        ?.steps?.[0]?.toolId;
+    } catch {
+      // Malformed pipeline JSON is reported when the definition is parsed below.
+    }
+    const inputModality = TOOLS.find((t) => t.id === firstToolId)?.modality ?? "image";
+    const pipelineScratch = join(tmpdir(), "snapotter-scratch", `pipeline-${randomUUID()}`);
+    await mkdir(pipelineScratch, { recursive: true });
+
+    if (inputModality === "image") {
       // Validate the initial image
       const validation = await validateImageBuffer(fileBuffer, filename);
       if (!validation.valid) {
@@ -279,6 +299,23 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
       } else {
         fileBuffer = await autoOrient(fileBuffer);
       }
+    } else {
+      // Non-image input: validate/decode via the tool's modality handler.
+      try {
+        const prepared = await inputHandlerFor(inputModality).prepare(fileBuffer, filename, {
+          scratchDir: pipelineScratch,
+        });
+        fileBuffer = prepared.buffer;
+        filename = prepared.filename;
+      } catch (err) {
+        if (err instanceof InputValidationError) {
+          const body: Record<string, string> = { error: err.message };
+          if (err.details) body.details = err.details;
+          return reply.status(err.statusCode).send(body);
+        }
+        throw err;
+      }
+    }
 
     // Parse and validate the pipeline definition
     if (!pipelineRaw) {
@@ -787,11 +824,20 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
     const preFailures: Array<{ originalIndex: number; filename: string; error: string }> = [];
     let flowChildIndex = 0;
 
+    // The first step's modality drives input validation for every file, so
+    // audio, video, and document pipelines are not rejected by the image
+    // validator.
+    const batchModality =
+      TOOLS.find((t) => t.id === pipeline.steps[0]?.toolId)?.modality ?? "image";
+    const pipelineBatchScratch = join(tmpdir(), "snapotter-scratch", `pipeline-batch-${parentId}`);
+    await mkdir(pipelineBatchScratch, { recursive: true });
+
     for (let fi = 0; fi < files.length; fi++) {
       const file = files[fi];
       let processBuffer = file.buffer;
       let processFilename = file.filename;
 
+      if (batchModality === "image") {
         const fileValidation = await validateImageBuffer(processBuffer, processFilename);
         if (!fileValidation.valid) {
           preFailures.push({
@@ -821,7 +867,11 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
         if (needsCliDecode(fileValidation.format)) {
           try {
             const fileExt = processFilename.split(".").pop()?.toLowerCase();
-          processBuffer = await decodeToSharpCompat(processBuffer, fileValidation.format, fileExt);
+            processBuffer = await decodeToSharpCompat(
+              processBuffer,
+              fileValidation.format,
+              fileExt,
+            );
             const ext = processFilename.match(/\.[^.]+$/)?.[0];
             if (ext) processFilename = `${processFilename.slice(0, -ext.length)}.png`;
           } catch {
@@ -834,6 +884,24 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
         } else {
           processBuffer = await autoOrient(processBuffer);
         }
+      } else {
+        // Non-image input: validate/decode via the tool's modality handler.
+        try {
+          const prepared = await inputHandlerFor(batchModality).prepare(
+            processBuffer,
+            processFilename,
+            { scratchDir: pipelineBatchScratch },
+          );
+          processBuffer = prepared.buffer;
+          processFilename = prepared.filename;
+        } catch (err) {
+          if (err instanceof InputValidationError) {
+            preFailures.push({ originalIndex: fi, filename: file.filename, error: err.message });
+            continue;
+          }
+          throw err;
+        }
+      }
 
       // Upload decoded file
       const perFileJobId = `${parentId}-f${flowChildIndex}`;