diff --git a/open-sse/services/batchProcessor.ts b/open-sse/services/batchProcessor.ts index 15a2a9b455..c0047a5ad0 100644 --- a/open-sse/services/batchProcessor.ts +++ b/open-sse/services/batchProcessor.ts @@ -58,19 +58,27 @@ export function stopBatchProcessor() { } /** - * Mark any in_progress batches as failed on startup. + * Mark any in_progress/finalizing batches as failed on startup. * These were orphaned by a server crash or restart and cannot be safely resumed. */ function recoverOrphanedBatches() { try { const pending = getPendingBatches(); for (const batch of pending) { - if (batch.status === "in_progress") { - console.warn(`[BATCH] Failing orphaned in_progress batch ${batch.id} (server restarted)`); + if (batch.status === "in_progress" || batch.status === "finalizing") { + const interruptedPhase = + batch.status === "finalizing" ? "during finalization" : "while processing requests"; + console.warn( + `[BATCH] Failing orphaned ${batch.status} batch ${batch.id} (server restarted)` + ); updateBatch(batch.id, { status: "failed", failedAt: Math.floor(Date.now() / 1000), - errors: [{ message: "Batch interrupted by server restart and cannot be resumed" }], + errors: [ + { + message: `Batch interrupted ${interruptedPhase} by server restart and cannot be resumed`, + }, + ], }); if (batch.inputFileId) { updateFileStatus(batch.inputFileId, "processed"); @@ -90,8 +98,8 @@ export async function processPendingBatches() { } else if (batch.status === "cancelling") { await cancelBatch(batch); } - // in_progress: currently being processed by processBatchItems running in background; - // orphaned in_progress batches are handled by recoverOrphanedBatches() at startup. + // in_progress/finalizing batches are either actively being worked by the current process + // or will be failed by recoverOrphanedBatches() on the next startup. } // Cleanup task: delete files for batches completed more than completionWindow ago diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 33e845b421..9bcedc2117 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1982,7 +1982,7 @@ async function handleRoundRobinCombo({ } } - if (isProviderBreakerOpenResponse(result, errorBody as any)) { + if (isProviderBreakerOpenResponse(result, errorBody as Record | null)) { lastError = errorText || String(result.status); if (!lastStatus) lastStatus = result.status; if (offset > 0) fallbackCount++; diff --git a/src/app/api/v1/_helpers/apiKeyScope.ts b/src/app/api/v1/_helpers/apiKeyScope.ts index 07c7afd8ac..cce0f76a75 100644 --- a/src/app/api/v1/_helpers/apiKeyScope.ts +++ b/src/app/api/v1/_helpers/apiKeyScope.ts @@ -40,6 +40,18 @@ export async function getApiKeyRequestScope(request: Request): Promise }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/batches/[id]/route.ts b/src/app/api/v1/batches/[id]/route.ts index c661fb3a21..e85aa336c4 100644 --- a/src/app/api/v1/batches/[id]/route.ts +++ b/src/app/api/v1/batches/[id]/route.ts @@ -1,4 +1,4 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getBatch } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; @@ -34,6 +34,10 @@ function formatBatchResponse(batch: any) { }; } +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/batches/route.ts b/src/app/api/v1/batches/route.ts index f732efb76d..39d24b619d 100644 --- a/src/app/api/v1/batches/route.ts +++ b/src/app/api/v1/batches/route.ts @@ -1,4 +1,4 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { createBatch, getFile, listBatches } from "@/lib/localDb"; import { v1BatchCreateSchema } from "@/shared/validation/schemas"; import { NextResponse } from "next/server"; @@ -35,6 +35,10 @@ function formatBatchResponse(batch: any) { }; } +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function POST(request: Request) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/[id]/content/route.ts b/src/app/api/v1/files/[id]/content/route.ts index 4cad037f60..70b1379af7 100644 --- a/src/app/api/v1/files/[id]/content/route.ts +++ b/src/app/api/v1/files/[id]/content/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getFile, getFileContent } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/[id]/route.ts b/src/app/api/v1/files/[id]/route.ts index 3b31db553f..5457349553 100644 --- a/src/app/api/v1/files/[id]/route.ts +++ b/src/app/api/v1/files/[id]/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getFile, deleteFile, formatFileResponse } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/route.ts b/src/app/api/v1/files/route.ts index cceb347f39..16f5b1945b 100644 --- a/src/app/api/v1/files/route.ts +++ b/src/app/api/v1/files/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { createFile, listFiles, formatFileResponse } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function POST(request: Request) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; @@ -38,16 +42,7 @@ export async function POST(request: Request) { const bytes = file.size; const filename = file.name; const mimeType = file.type; - - // Stream the upload into memory in chunks to avoid allocating a large contiguous ArrayBuffer - const reader = file.stream().getReader(); - const chunks: Uint8Array[] = []; - while (true) { - const { done, value } = await reader.read(); - if (done) break; - chunks.push(value); - } - const content = Buffer.concat(chunks.map((c) => Buffer.from(c))); + const content = Buffer.from(await file.arrayBuffer()); let expiresAt: number | undefined; if (expiresAfterAnchor === "created_at" && expiresAfterSeconds) { diff --git a/src/lib/db/batches.ts b/src/lib/db/batches.ts index f96dcd99f3..71515f0a7a 100644 --- a/src/lib/db/batches.ts +++ b/src/lib/db/batches.ts @@ -171,7 +171,9 @@ export function listBatches(apiKeyId?: string, limit: number = 20, after?: strin export function getPendingBatches(): BatchRecord[] { const db = getDbInstance(); const rows = db - .prepare("SELECT * FROM batches WHERE status IN ('validating', 'in_progress', 'cancelling')") + .prepare( + "SELECT * FROM batches WHERE status IN ('validating', 'in_progress', 'finalizing', 'cancelling')" + ) .all(); return rows.map((row) => parseBatchRow(row)); } diff --git a/src/lib/guardrails/visionBridgeHelpers.ts b/src/lib/guardrails/visionBridgeHelpers.ts index c0d28376b6..d82cdb334e 100644 --- a/src/lib/guardrails/visionBridgeHelpers.ts +++ b/src/lib/guardrails/visionBridgeHelpers.ts @@ -110,6 +110,34 @@ export function resolveImageAsDataUri(imageUrl: string): string { return `data:image/png;base64,${imageUrl}`; } +async function fetchRemoteImageAsDataUri(imageUrl: string, signal: AbortSignal): Promise { + const response = await fetch(imageUrl, { signal }); + if (!response.ok) { + throw new Error(`Vision image fetch error ${response.status}`); + } + + const mediaType = response.headers.get("content-type")?.split(";")[0]?.trim() || "image/png"; + const bytes = Buffer.from(await response.arrayBuffer()); + return `data:${mediaType};base64,${bytes.toString("base64")}`; +} + +async function normalizeVisionImageInput( + imageInput: string, + isAnthropic: boolean, + signal: AbortSignal +): Promise { + const normalizedImage = resolveImageAsDataUri(imageInput); + + if ( + isAnthropic && + (normalizedImage.startsWith("http://") || normalizedImage.startsWith("https://")) + ) { + return fetchRemoteImageAsDataUri(normalizedImage, signal); + } + + return normalizedImage; +} + export interface VisionModelConfig { model: string; prompt: string; @@ -137,9 +165,12 @@ export async function callVisionModel( try { // Extract model name from provider/model format - const modelName = config.model.includes("/") - ? config.model.split("/")[1] - : config.model; + const modelName = config.model.includes("/") ? config.model.split("/")[1] : config.model; + const normalizedImageInput = await normalizeVisionImageInput( + imageDataUri, + isAnthropic, + controller.signal + ); let response: Response; @@ -148,9 +179,9 @@ export async function callVisionModel( const anthropicBaseUrl = process.env.ANTHROPIC_API_URL || "https://api.anthropic.com"; // Parse data URI to extract media type and base64 data - const matches = imageDataUri.match(/^data:([^;]+);base64,(.+)$/); + const matches = normalizedImageInput.match(/^data:([^;]+);base64,(.+)$/); let mediaType = "image/png"; - let base64Data = imageDataUri; + let base64Data = normalizedImageInput; if (matches) { mediaType = matches[1]; @@ -209,7 +240,7 @@ export async function callVisionModel( { type: "image_url", image_url: { - url: imageDataUri, + url: normalizedImageInput, detail: "low", }, }, @@ -239,7 +270,9 @@ export async function callVisionModel( }; if (anthropicData.error) { - throw new Error(`Vision API error: ${anthropicData.error.message || JSON.stringify(anthropicData.error)}`); + throw new Error( + `Vision API error: ${anthropicData.error.message || JSON.stringify(anthropicData.error)}` + ); } const textContent = anthropicData.content?.find((c) => c.type === "text"); @@ -257,7 +290,9 @@ export async function callVisionModel( }; if (openaiData.error) { - throw new Error(`Vision API error: ${openaiData.error.message || JSON.stringify(openaiData.error)}`); + throw new Error( + `Vision API error: ${openaiData.error.message || JSON.stringify(openaiData.error)}` + ); } const content = openaiData.choices?.[0]?.message?.content; diff --git a/tests/unit/batch_api.test.ts b/tests/unit/batch_api.test.ts index 808dacc5cb..5a4a323e84 100644 --- a/tests/unit/batch_api.test.ts +++ b/tests/unit/batch_api.test.ts @@ -6,6 +6,7 @@ import path from "node:path"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-batch-api-")); process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = process.env.API_KEY_SECRET || "test-secret-123"; const { createFile, @@ -25,7 +26,11 @@ const { getDbInstance } = await import("../../src/lib/db/core.ts"); const { initBatchProcessor, stopBatchProcessor, processPendingBatches } = await import("../../open-sse/services/batchProcessor.ts"); const batchesRoute = await import("../../src/app/api/v1/batches/route.ts"); +const batchByIdRoute = await import("../../src/app/api/v1/batches/[id]/route.ts"); +const batchCancelRoute = await import("../../src/app/api/v1/batches/[id]/cancel/route.ts"); const filesRoute = await import("../../src/app/api/v1/files/route.ts"); +const fileByIdRoute = await import("../../src/app/api/v1/files/[id]/route.ts"); +const fileContentRoute = await import("../../src/app/api/v1/files/[id]/content/route.ts"); test("Batch API and Processing", async () => { // 0. Setup environment, mock provider and API key @@ -654,6 +659,43 @@ test("Batch cleanup honors output_expires_after for output artifacts", async () assert.equal(getFile(errorFile.id), null); }); +test("Batch processor fails orphaned finalizing batches during startup recovery", async () => { + const apiKey = await createApiKey("Finalizing Recovery Key", "test-machine"); + const inputFile = createFile({ + bytes: 2, + filename: "finalizing.jsonl", + purpose: "batch", + content: Buffer.from("{}"), + apiKeyId: apiKey.id, + }); + + const batch = createBatch({ + endpoint: "/v1/chat/completions", + completionWindow: "24h", + inputFileId: inputFile.id, + apiKeyId: apiKey.id, + }); + + updateBatch(batch.id, { + status: "finalizing", + finalizingAt: Math.floor(Date.now() / 1000), + }); + + initBatchProcessor(); + + try { + const recoveredBatch = getBatch(batch.id); + assert.strictEqual(recoveredBatch?.status, "failed"); + assert.match( + String(recoveredBatch?.errors?.[0]?.message || ""), + /interrupted during finalization/i + ); + assert.strictEqual(getFile(inputFile.id)?.status, "processed"); + } finally { + stopBatchProcessor(); + } +}); + test("Batch list route rejects missing API key when REQUIRE_API_KEY is enabled", async () => { const previous = process.env.REQUIRE_API_KEY; process.env.REQUIRE_API_KEY = "true"; @@ -688,6 +730,78 @@ test("Files list route rejects invalid API key when REQUIRE_API_KEY is enabled", } }); +test("Files upload route rejects invalid API key even when auth is optional", async () => { + const previous = process.env.REQUIRE_API_KEY; + process.env.REQUIRE_API_KEY = "false"; + + try { + const formData = new FormData(); + formData.set("purpose", "batch"); + formData.set( + "file", + new File([Buffer.from('{"ok":true}\n')], "input.jsonl", { type: "application/json" }) + ); + + const response = await filesRoute.POST( + new Request("http://localhost/api/v1/files", { + method: "POST", + headers: { Authorization: "Bearer invalid-test-key" }, + body: formData, + }) + ); + const json = await response.json(); + + assert.strictEqual(response.status, 401); + assert.strictEqual(json.error.message, "Invalid API key"); + } finally { + process.env.REQUIRE_API_KEY = previous ?? "false"; + } +}); + +test("Files upload route stores multipart content", async () => { + const fileContent = '{"custom_id":"req-1"}\n'; + const formData = new FormData(); + formData.set("purpose", "batch"); + formData.set( + "file", + new File([Buffer.from(fileContent)], "upload.jsonl", { type: "application/json" }) + ); + + const response = await filesRoute.POST( + new Request("http://localhost/api/v1/files", { + method: "POST", + body: formData, + }) + ); + const json = await response.json(); + + assert.strictEqual(response.status, 200); + assert.ok(json.id); + assert.strictEqual(getFileContent(json.id)?.toString(), fileContent); +}); + +test("Files and batches routes expose explicit CORS preflight handlers", async () => { + const routes = [ + batchesRoute, + batchByIdRoute, + batchCancelRoute, + filesRoute, + fileByIdRoute, + fileContentRoute, + ]; + + for (const route of routes) { + assert.strictEqual(typeof route.OPTIONS, "function"); + const response = await route.OPTIONS(); + assert.strictEqual(response.status, 204); + assert.strictEqual(response.headers.get("Access-Control-Allow-Origin"), "*"); + assert.match( + String(response.headers.get("Access-Control-Allow-Headers") || ""), + /Authorization/i + ); + } +}); + test("Batch Cancel API", async () => { const apiKey = await createApiKey("Cancel Test Key", "test-machine"); diff --git a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts index dd241214e1..ed98e35449 100644 --- a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts +++ b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts @@ -56,8 +56,7 @@ test("callVisionModel throws on HTTP error", async () => { }; await assert.rejects( - async () => - await callVisionModel("data:image/png;base64,iVBORw0KGgo", config), + async () => await callVisionModel("data:image/png;base64,iVBORw0KGgo", config), /Vision API error 500/ ); } finally { @@ -168,10 +167,7 @@ test("callVisionModel passes custom API key", async () => { await callVisionModel("data:image/png;base64,iVBORw0KGgo", config, "sk-custom-key"); - assert.strictEqual( - capturedHeaders["Authorization"], - "Bearer sk-custom-key" - ); + assert.strictEqual(capturedHeaders["Authorization"], "Bearer sk-custom-key"); } finally { globalThis.fetch = originalFetch; } @@ -210,22 +206,75 @@ test("callVisionModel uses correct request body format", async () => { assert.ok(Array.isArray(capturedBody.messages)); assert.strictEqual((capturedBody.messages as unknown[]).length, 1); - const message = (capturedBody.messages as Array<{role: string; content: unknown[]}>)[0]; + const message = (capturedBody.messages as Array<{ role: string; content: unknown[] }>)[0]; assert.strictEqual(message.role, "user"); assert.ok(Array.isArray(message.content)); assert.strictEqual(message.content.length, 2); // First content is image_url - const imagePart = message.content[0] as {type: string; image_url: {url: string; detail: string}}; + const imagePart = message.content[0] as { + type: string; + image_url: { url: string; detail: string }; + }; assert.strictEqual(imagePart.type, "image_url"); assert.strictEqual(imagePart.image_url.url, imageUri); assert.strictEqual(imagePart.image_url.detail, "low"); // Second content is text prompt - const textPart = message.content[1] as {type: string; text: string}; + const textPart = message.content[1] as { type: string; text: string }; assert.strictEqual(textPart.type, "text"); assert.strictEqual(textPart.text, "What is in this image?"); } finally { globalThis.fetch = originalFetch; } }); + +test("callVisionModel fetches remote images before Anthropic requests", async () => { + const fetchCalls: Array<{ url: string; init?: RequestInit }> = []; + + globalThis.fetch = async (url: URL | RequestInfo, init?: RequestInit) => { + const requestUrl = String(url); + fetchCalls.push({ url: requestUrl, init }); + + if (requestUrl === "https://cdn.example.com/cat.png") { + return new Response(Buffer.from("cat-image-bytes"), { + status: 200, + headers: { "Content-Type": "image/png" }, + }); + } + + return new Response( + JSON.stringify({ + content: [{ type: "text", text: "A cat sitting on a chair" }], + }), + { + status: 200, + headers: { "Content-Type": "application/json" }, + } + ); + }; + + try { + const config: VisionModelConfig = { + model: "anthropic/claude-3-haiku", + prompt: "Describe this image", + timeoutMs: 30000, + maxImages: 10, + }; + + const result = await callVisionModel("https://cdn.example.com/cat.png", config, "sk-ant"); + + assert.strictEqual(result, "A cat sitting on a chair"); + assert.strictEqual(fetchCalls.length, 2); + assert.strictEqual(fetchCalls[0].url, "https://cdn.example.com/cat.png"); + assert.strictEqual(fetchCalls[1].url, "https://api.anthropic.com/v1/messages"); + + const anthropicBody = JSON.parse(fetchCalls[1].init?.body as string); + const imageSource = anthropicBody.messages[0].content[0].source; + assert.strictEqual(imageSource.type, "base64"); + assert.strictEqual(imageSource.media_type, "image/png"); + assert.strictEqual(imageSource.data, Buffer.from("cat-image-bytes").toString("base64")); + } finally { + globalThis.fetch = originalFetch; + } +});