fix: add batch item dispatching to specific handlers based on URL (#1495)

Integrated into release/v3.7.0

open-sse/services/batchProcessor.ts

@@ -1,17 +1,21 @@
 import { v4 as uuidv4 } from "uuid";
 import {
-  getPendingBatches,
-  getTerminalBatches,
-  updateBatch,
-  getFileContent,
   createFile,
+  deleteFile,
   getApiKeyById,
   getBatch,
+  getFileContent,
+  getPendingBatches,
+  getTerminalBatches,
   listFiles,
-  deleteFile,
+  updateBatch,
   updateFileStatus,
 } from "@/lib/localDb";
 import { handleChat } from "@/sse/handlers/chat";
+import { POST as handleEmbeddingsRoute } from "@/app/api/v1/embeddings/route";
+import { POST as handleModerationsRoute } from "@/app/api/v1/moderations/route";
+import { POST as handleImagesGenerationsRoute } from "@/app/api/v1/images/generations/route";
+import { POST as handleVideosGenerationsRoute } from "@/app/api/v1/videos/generations/route";
 
 let isProcessing = false;
 let pollInterval: NodeJS.Timeout | null = null;
@@ -175,6 +179,49 @@ async function startBatch(batch: any) {
   }
 }
 
+/**
+ * Dispatch a single batch item to the correct handler based on the URL.
+ * Each batch item may target a different API endpoint (chat, embeddings, etc.).
+ */
+async function dispatchBatchItem(
+  url: string,
+  headers: Headers,
+  body: Record<string, unknown>
+): Promise<Response> {
+  const request = new Request(`http://localhost${url}`, {
+    method: "POST",
+    headers,
+    body: JSON.stringify(body),
+  });
+
+  switch (url) {
+    case "/v1/embeddings":
+      return handleEmbeddingsRoute(request);
+    case "/v1/moderations":
+      return handleModerationsRoute(request);
+    case "/v1/images/generations":
+    case "/v1/images/edits":
+      return handleImagesGenerationsRoute(request);
+    case "/v1/videos":
+    case "/v1/videos/generations":
+      return handleVideosGenerationsRoute(request);
+      // /v1/chat/completions, /v1/completions, /v1/responses
+    default: {
+      return handleChat(
+        new Request(`http://localhost${url}`, {
+          method: "POST",
+          headers,
+          body: JSON.stringify({
+            ...body,
+            // BATCH-SPECIFIC: Force stream: false — batches don't support SSE responses
+            stream: body.stream ?? false,
+          }),
+        })
+      );
+    }
+  }
+}
+
 async function processBatchItems(batch: any, lines: string[]) {
   const results: any[] = [];
   const errors: any[] = [];
@@ -204,15 +251,7 @@ async function processBatchItems(batch: any, lines: string[]) {
       }
       headers.set("Content-Type", "application/json");
 
-      // BATCH-SPECIFIC: Force stream: false — batches don't support SSE responses
-      const batchItemBody = { ...body, stream: false };
-
-      const response = await handleChat({
-        json: async () => batchItemBody,
-        url: `http://localhost${url}`,
-        headers,
-        method: "POST",
-      } as any);
+      const response = await dispatchBatchItem(url, headers, body);
 
       let responseData: { error: any; id?: any; usage?: any; model?: any };
       let statusCode = 200;

tests/manual/batch.http

@@ -21,6 +21,35 @@ batch
     client.global.set("file_id", response.body.id);
 %}
 
+###
+# @name Upload a JSONL file for batch processing - test embedding endpoint
+POST {{omniroute-address}}/v1/files
+Authorization: Bearer {{OMNIROUTE_API_KEY}}
+Content-Type: multipart/form-data; boundary=boundary
+
+--boundary
+Content-Disposition: form-data; name="file"; filename="batch_input.jsonl"
+Content-Type: application/jsonl
+
+{"custom_id": "request-1", "method": "POST", "url": "/v1/embeddings", "body": {"model": "pinecone/llama-text-embed-v2", "input": "The food was delicious and the waiter was very attentive."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "The food was delicious and the waiter was very attentive."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "I had a terrible experience at the restaurant. The food was cold and the service was slow."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Too bad! The food was cold and the service was slow."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Best dining experience ever! The food was amazing and the waiter was super friendly."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "All you can eat buffet was a disaster. The food was stale and the staff was rude."}}
+{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "All your base are belong to us."}}
+
+--boundary
+Content-Disposition: form-data; name="purpose"
+
+batch
+--boundary--
+
+> {%
+    client.global.set("file_id", response.body.id);
+%}
+
 ###
 # @name List all uploaded files
 GET {{omniroute-address}}/v1/files

tests/unit/batch_api.test.ts

@@ -703,6 +703,65 @@ test("Retrieve file content spec compliance", async () => {
   assert.ok(unauthorized);
 });
 
+test("Batch dispatches to embeddings handler for /v1/embeddings URL", async () => {
+    initBatchProcessor();
+    try {
+        const batchItems = [
+            JSON.stringify({
+                custom_id: "embed-request-1",
+                method: "POST",
+                url: "/v1/embeddings",
+                body: { model: "mistral/mistral-embed", input: "The food was delicious." }
+            })
+        ].join("\n");
+
+        const file = createFile({
+            bytes: Buffer.byteLength(batchItems),
+            filename: "embed_batch.jsonl",
+            purpose: "batch",
+            content: Buffer.from(batchItems),
+            apiKeyId: null
+        });
+
+        const batch = createBatch({
+            endpoint: "/v1/embeddings",
+            completionWindow: "24h",
+            inputFileId: file.id,
+            apiKeyId: null
+        });
+
+        let maxAttempts = 20;
+        let currentBatch = getBatch(batch.id);
+        while (maxAttempts > 0 && currentBatch?.status !== "completed" && currentBatch?.status !== "failed") {
+            await new Promise(resolve => setTimeout(resolve, 1000));
+            currentBatch = getBatch(batch.id);
+            maxAttempts--;
+        }
+
+        assert.ok(
+            currentBatch?.status === "completed" || currentBatch?.status === "failed",
+            "Batch should reach a terminal state"
+        );
+        assert.strictEqual(currentBatch?.requestCountsTotal, 1);
+
+        // Verify the batch item was dispatched to the embeddings handler, not the chat handler.
+        // The chat handler would return errors about missing "messages", "Missing model", etc.
+        // The embeddings handler returns errors about missing credentials or invalid embedding models.
+        const outputFileId = currentBatch?.outputFileId || currentBatch?.errorFileId;
+        assert.ok(outputFileId, "Should have an output or error file");
+        const outputContent = getFileContent(outputFileId!);
+        assert.ok(outputContent, "Output file should have content");
+        const result = JSON.parse(outputContent.toString());
+        const errorMsg = result.response?.body?.error?.message || "";
+        assert.ok(
+            !errorMsg.includes("messages") && !errorMsg.includes("Missing model"),
+            `Error should not be a chat-specific error. Got: ${errorMsg}`
+        );
+    } finally {
+        stopBatchProcessor();
+    }
+});
+
 test("getTerminalBatches returns only terminal statuses ordered oldest first", async () => {
   const apiKey = await createApiKey("Terminal Batches Test Key", "test-machine");