Croway commented on code in PR #26601: URL: https://github.com/apache/camel/pull/26601#discussion_r4048117625
########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java: ########## @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a + * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route + * produced. + */ +public class BatchStore { + + /** + * A file uploaded by a route or generated by the mock for a finished batch. + */ + public record StoredFile(String id, String filename, String purpose, byte[] content) { + } + + /** + * A created batch and the position it reached in the configured status progression. + */ + public static class StoredBatch { + final String id; + final String inputFileId; + final String endpoint; + final String metadata; + int statusIndex; + boolean cancelling; + boolean cancelled; + boolean resultsBuilt; Review Comment: The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is. _Claude Code on behalf of Federico Mariani (Croway)_ ########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java: ########## @@ -0,0 +1,404 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Pattern; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.sun.net.httpserver.HttpExchange; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving + * the result files. + * <p> + * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence + * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next + * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input + * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}. + */ +public class BatchRequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class); + + static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed"); + + private static final String DEFAULT_RESPONSE_BODY + = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\"," + + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"}," + + "\"finish_reason\":\"stop\"}]}"; + + private final List<BatchExpectation> expectations; + private final List<String> statuses; + private final BatchStore store; + private final ObjectMapper objectMapper; + + public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store, + ObjectMapper objectMapper) { + this.expectations = expectations; + this.statuses = statuses; + this.store = store; + this.objectMapper = objectMapper; + } + + /** + * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the + * handlers of the inference APIs. + */ + public static boolean handles(String path) { + return path.endsWith("/files") || path.contains("/files/") + || path.endsWith("/batches") || path.contains("/batches/"); + } + + public void handleRequest(HttpExchange exchange) throws IOException { + String path = exchange.getRequestURI().getPath(); + String method = exchange.getRequestMethod(); + try { + if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) { + uploadFile(exchange); + } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) { + fileContent(exchange, idBefore(path, "/content")); + } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) { + fileObject(exchange, lastSegment(path)); + } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) { + deleteFile(exchange, lastSegment(path)); + } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) { + createBatch(exchange); + } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) { + cancelBatch(exchange, idBefore(path, "/cancel")); + } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) { + retrieveBatch(exchange, lastSegment(path)); + } else { + sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path); + } + } catch (Exception e) { + LOG.error("Error processing batch request {} {}", method, path, e); + sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage())); + } + } + + private void uploadFile(HttpExchange exchange) throws IOException { + byte[] body; + try (InputStream is = exchange.getRequestBody()) { + body = is.readAllBytes(); + } + String contentType = exchange.getRequestHeaders().getFirst("Content-Type"); + String boundary = contentType != null && contentType.contains("boundary=") + ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null; + if (boundary == null) { + sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request"); + return; + } + + // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged + String raw = new String(body, StandardCharsets.ISO_8859_1); + String filename = "unknown"; + String purpose = "unknown"; + byte[] content = new byte[0]; + // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern + for (String part : raw.split(Pattern.quote("--" + boundary))) { + int headerEnd = part.indexOf("\r\n\r\n"); + if (headerEnd < 0) { + continue; + } + String headers = part.substring(0, headerEnd); + String value = part.substring(headerEnd + 4); + if (value.endsWith("\r\n")) { + value = value.substring(0, value.length() - 2); + } + if (headers.contains("name=\"file\"")) { + filename = between(headers, "filename=\"", "\""); + content = value.getBytes(StandardCharsets.ISO_8859_1); + } else if (headers.contains("name=\"purpose\"")) { + purpose = value.trim(); + } + } + + String id = store.addFile(filename, purpose, content); + LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose); + sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString()); + } + + private void fileContent(HttpExchange exchange, String fileId) throws IOException { + byte[] content = store.getFileContent(fileId); + if (content == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + exchange.getResponseHeaders().add("Content-Type", "application/octet-stream"); + exchange.sendResponseHeaders(200, content.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(content); + } + } + + private void fileObject(HttpExchange exchange, String fileId) throws IOException { + BatchStore.StoredFile file = store.file(fileId); + if (file == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString()); + } + + private void deleteFile(HttpExchange exchange, String fileId) throws IOException { + if (store.file(fileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + store.removeFile(fileId); + ObjectNode node = objectMapper.createObjectNode(); + node.put("id", fileId); + node.put("object", "file"); + node.put("deleted", true); + sendJson(exchange, 200, node.toString()); + } + + private void createBatch(HttpExchange exchange) throws IOException { + JsonNode request; + try (InputStream is = exchange.getRequestBody()) { + request = objectMapper.readTree(is.readAllBytes()); + } + String inputFileId = request.path("input_file_id").asText(); + if (store.file(inputFileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId); + return; + } + JsonNode metadata = request.get("metadata"); + BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(), + metadata != null && !metadata.isNull() ? metadata.toString() : null); + maybeBuildResults(batch); + LOG.debug("Created batch {} for input file {}", batch.id, inputFileId); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (batch.cancelling) { + // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does + batch.cancelling = false; + batch.cancelled = true; + } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) { + batch.statusIndex++; + } + maybeBuildResults(batch); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void cancelBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (isFinal(status(batch))) { + sendError(exchange, 400, "invalid_request_error", + "Cannot cancel a batch with status '" + status(batch) + "'."); + return; + } + batch.cancelling = true; + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private List<String> statusList() { + return statuses.isEmpty() ? DEFAULT_STATUSES : statuses; + } + + private String status(BatchStore.StoredBatch batch) { + if (batch.cancelled) { + return "cancelled"; + } + if (batch.cancelling) { + return "cancelling"; + } + return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1)); + } + + private boolean isFinal(String status) { + return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status); + } + + /** + * Builds the output and error files once the batch reaches a final status, as the API does when it finishes + * processing the input file. + */ + private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException { + if (batch.resultsBuilt || !isFinal(status(batch))) { + return; + } + batch.resultsBuilt = true; + + BatchStore.StoredFile input = store.file(batch.inputFileId); + List<String> outputLines = new ArrayList<>(); + List<String> errorLines = new ArrayList<>(); + int line = 0; + for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) { Review Comment: Applied in d2a51a238e8b: the split is now `\\r?\\n`. The `newLine()` trigger no longer exists (lines are joined with an explicit `\\n` since ddafa304), but a raw JSONL body produced on Windows still can carry CRLF. _Claude Code on behalf of Federico Mariani (Croway)_ ########## components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchResultsProducer.java: ########## @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai; + +import java.util.Optional; + +import com.openai.core.http.HttpResponse; +import com.openai.models.batches.Batch; +import org.apache.camel.CamelExchangeException; +import org.apache.camel.Exchange; +import org.apache.camel.component.ai.observability.GenAiErrorSupport; +import org.apache.camel.support.DefaultProducer; +import org.apache.camel.support.SynchronizationAdapter; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * OpenAI producer for the {@code batch-results} operation, which downloads the output or error file of a finished + * batch. + * <p> + * The body is the file as an {@link java.io.InputStream}, so a large result can be split line by line with the splitter + * in streaming mode. The HTTP response is closed when the exchange completes. Stream caching reads such a body into + * memory before the next step, so a route handling large results should turn it off. + */ +public class OpenAIBatchResultsProducer extends DefaultProducer { + + private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchResultsProducer.class); + private static final String ERROR_FILE = "error"; + + public OpenAIBatchResultsProducer(OpenAIEndpoint endpoint) { + super(endpoint); + } + + @Override + public OpenAIEndpoint getEndpoint() { + return (OpenAIEndpoint) super.getEndpoint(); + } + + @Override + public void process(Exchange exchange) throws Exception { + OpenAIConfiguration config = getEndpoint().getConfiguration(); + String batchId = exchange.getIn().getHeader(OpenAIConstants.BATCH_ID, String.class); + if (ObjectHelper.isEmpty(batchId)) { + throw new IllegalArgumentException( + "The " + OpenAIConstants.BATCH_ID + " header must hold the id of the batch to read the results of"); + } + + String resultsFile = exchange.getIn().getHeader(OpenAIConstants.BATCH_RESULTS_FILE, + config.getBatchResultsFile(), String.class); + if (!"output".equals(resultsFile) && !ERROR_FILE.equals(resultsFile)) { + throw new IllegalArgumentException( + "Unsupported batch results file: " + resultsFile + ". Supported: output, error"); + } + + Batch batch; + try { + batch = getEndpoint().getClient().batches().retrieve(batchId); + } catch (RuntimeException e) { + GenAiErrorSupport.apply(exchange, e); + throw e; + } + OpenAIBatchSupport.setBatchHeaders(exchange.getMessage(), batch); + + // the API sets the file ids once it has stopped processing, so their presence, not the status, says whether + // the results can be read; the status only explains why a file is missing + Optional<String> fileId = ERROR_FILE.equals(resultsFile) ? batch.errorFileId() : batch.outputFileId(); + if (fileId.isEmpty()) { + switch (batch.status().value()) { + case FAILED -> + // the input was rejected, so no result file exists: the reason is in the errors of the batch + throw new CamelExchangeException( + "Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange); Review Comment: Applied in d2a51a238e8b: a failed batch with no `errors` now reports "the API reported no validation errors" instead of `[]`. Worded neutrally since the component also targets compatible servers. _Claude Code on behalf of Federico Mariani (Croway)_ ########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java: ########## @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a + * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route + * produced. + */ +public class BatchStore { + + /** + * A file uploaded by a route or generated by the mock for a finished batch. + */ + public record StoredFile(String id, String filename, String purpose, byte[] content) { + } + + /** + * A created batch and the position it reached in the configured status progression. + */ + public static class StoredBatch { + final String id; + final String inputFileId; + final String endpoint; + final String metadata; + int statusIndex; + boolean cancelling; + boolean cancelled; + boolean resultsBuilt; + String outputFileId; + String errorFileId; + int total; + int completed; + int failed; Review Comment: The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is. _Claude Code on behalf of Federico Mariani (Croway)_ ########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java: ########## @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a + * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route + * produced. + */ +public class BatchStore { + + /** + * A file uploaded by a route or generated by the mock for a finished batch. + */ + public record StoredFile(String id, String filename, String purpose, byte[] content) { + } + + /** + * A created batch and the position it reached in the configured status progression. + */ + public static class StoredBatch { + final String id; + final String inputFileId; + final String endpoint; + final String metadata; + int statusIndex; + boolean cancelling; + boolean cancelled; + boolean resultsBuilt; + String outputFileId; + String errorFileId; + int total; + int completed; + int failed; + Review Comment: The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is. _Claude Code on behalf of Federico Mariani (Croway)_ ########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java: ########## @@ -0,0 +1,404 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Pattern; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.sun.net.httpserver.HttpExchange; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving + * the result files. + * <p> + * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence + * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next + * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input + * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}. + */ +public class BatchRequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class); + + static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed"); + + private static final String DEFAULT_RESPONSE_BODY + = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\"," + + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"}," + + "\"finish_reason\":\"stop\"}]}"; + + private final List<BatchExpectation> expectations; + private final List<String> statuses; + private final BatchStore store; + private final ObjectMapper objectMapper; + + public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store, + ObjectMapper objectMapper) { + this.expectations = expectations; + this.statuses = statuses; + this.store = store; + this.objectMapper = objectMapper; + } + + /** + * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the + * handlers of the inference APIs. + */ + public static boolean handles(String path) { + return path.endsWith("/files") || path.contains("/files/") + || path.endsWith("/batches") || path.contains("/batches/"); + } + + public void handleRequest(HttpExchange exchange) throws IOException { + String path = exchange.getRequestURI().getPath(); + String method = exchange.getRequestMethod(); + try { + if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) { + uploadFile(exchange); + } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) { + fileContent(exchange, idBefore(path, "/content")); + } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) { + fileObject(exchange, lastSegment(path)); + } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) { + deleteFile(exchange, lastSegment(path)); + } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) { + createBatch(exchange); + } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) { + cancelBatch(exchange, idBefore(path, "/cancel")); + } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) { + retrieveBatch(exchange, lastSegment(path)); + } else { + sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path); + } + } catch (Exception e) { + LOG.error("Error processing batch request {} {}", method, path, e); + sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage())); + } + } + + private void uploadFile(HttpExchange exchange) throws IOException { + byte[] body; + try (InputStream is = exchange.getRequestBody()) { + body = is.readAllBytes(); + } + String contentType = exchange.getRequestHeaders().getFirst("Content-Type"); + String boundary = contentType != null && contentType.contains("boundary=") + ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null; + if (boundary == null) { + sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request"); + return; + } + + // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged + String raw = new String(body, StandardCharsets.ISO_8859_1); + String filename = "unknown"; + String purpose = "unknown"; + byte[] content = new byte[0]; + // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern + for (String part : raw.split(Pattern.quote("--" + boundary))) { + int headerEnd = part.indexOf("\r\n\r\n"); + if (headerEnd < 0) { + continue; + } + String headers = part.substring(0, headerEnd); + String value = part.substring(headerEnd + 4); + if (value.endsWith("\r\n")) { + value = value.substring(0, value.length() - 2); + } + if (headers.contains("name=\"file\"")) { + filename = between(headers, "filename=\"", "\""); + content = value.getBytes(StandardCharsets.ISO_8859_1); + } else if (headers.contains("name=\"purpose\"")) { + purpose = value.trim(); + } + } + + String id = store.addFile(filename, purpose, content); + LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose); + sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString()); + } + + private void fileContent(HttpExchange exchange, String fileId) throws IOException { + byte[] content = store.getFileContent(fileId); + if (content == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + exchange.getResponseHeaders().add("Content-Type", "application/octet-stream"); + exchange.sendResponseHeaders(200, content.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(content); + } + } + + private void fileObject(HttpExchange exchange, String fileId) throws IOException { + BatchStore.StoredFile file = store.file(fileId); + if (file == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString()); + } + + private void deleteFile(HttpExchange exchange, String fileId) throws IOException { + if (store.file(fileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + store.removeFile(fileId); + ObjectNode node = objectMapper.createObjectNode(); + node.put("id", fileId); + node.put("object", "file"); + node.put("deleted", true); + sendJson(exchange, 200, node.toString()); + } + + private void createBatch(HttpExchange exchange) throws IOException { + JsonNode request; + try (InputStream is = exchange.getRequestBody()) { + request = objectMapper.readTree(is.readAllBytes()); + } + String inputFileId = request.path("input_file_id").asText(); + if (store.file(inputFileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId); + return; + } + JsonNode metadata = request.get("metadata"); + BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(), + metadata != null && !metadata.isNull() ? metadata.toString() : null); + maybeBuildResults(batch); + LOG.debug("Created batch {} for input file {}", batch.id, inputFileId); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (batch.cancelling) { + // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does + batch.cancelling = false; + batch.cancelled = true; + } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) { + batch.statusIndex++; + } + maybeBuildResults(batch); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void cancelBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (isFinal(status(batch))) { + sendError(exchange, 400, "invalid_request_error", + "Cannot cancel a batch with status '" + status(batch) + "'."); + return; + } + batch.cancelling = true; + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private List<String> statusList() { + return statuses.isEmpty() ? DEFAULT_STATUSES : statuses; + } + + private String status(BatchStore.StoredBatch batch) { + if (batch.cancelled) { + return "cancelled"; + } + if (batch.cancelling) { + return "cancelling"; + } + return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1)); + } + + private boolean isFinal(String status) { + return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status); + } + + /** + * Builds the output and error files once the batch reaches a final status, as the API does when it finishes + * processing the input file. + */ + private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException { + if (batch.resultsBuilt || !isFinal(status(batch))) { + return; + } + batch.resultsBuilt = true; + + BatchStore.StoredFile input = store.file(batch.inputFileId); + List<String> outputLines = new ArrayList<>(); + List<String> errorLines = new ArrayList<>(); + int line = 0; + for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) { Review Comment: Applied in d2a51a238e8b: the split is now `\\r?\\n`. The `newLine()` trigger no longer exists (lines are joined with an explicit `\\n` since ddafa304), but a raw JSONL body produced on Windows still can carry CRLF. _Claude Code on behalf of Federico Mariani (Croway)_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
