CAMEL-24559: Address gnodet review feedback on OpenAI observability

- Pass OpenAI token counts as Long without Math.toIntExact (includes
  CAMEL-24560 GenAiUsage Long token fields cherry-picked from follow-up)
- Extract observedCall helper in OpenAIResponsesProducer to deduplicate
  observation boilerplate between createResponse and createStructuredResponse
- Restore moderation security comments explaining why mismatched result
  counts must fail the exchange

Co-authored-by: Cursor Agent <noreply@cursor.com>

Author: cursoragent

components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEmbeddingsProducer.java

@@ -117,7 +117,7 @@ private void processInternal(Exchange exchange) throws Exception {
             response = getEndpoint().getClient().embeddings().create(params);
             var usage = response.usage();
             observation.recordSuccess(GenAiUsage.of(
-                    usage != null ? toTokenCount(usage.promptTokens()) : null,
+                    usage != null ? usage.promptTokens() : null,
                     null,
                     null,
                     response.model()));
@@ -151,10 +151,6 @@ private void processInternal(Exchange exchange) throws Exception {
         calculateSimilarityIfRequested(exchange, embeddings);
     }
 
-    private static Integer toTokenCount(long tokens) {
-        return Math.toIntExact(tokens);
-    }
-
     @SuppressWarnings("unchecked")
     private List<String> extractInputs(Message in) throws Exception {
         Object body = in.getBody();

components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIModerationProducer.java

@@ -103,7 +103,7 @@ private void processInternal(Exchange exchange) throws Exception {
         ModerationCreateResponse response;
         try {
             response = getEndpoint().getClient().moderations().create(paramsBuilder.build());
-            observation.recordSuccess(GenAiUsage.of(null, null, null, response.model()));
+            observation.recordSuccess(GenAiUsage.of((Long) null, null, null, response.model()));
         } catch (Exception e) {
             GenAiErrorSupport.apply(exchange, e);
             observation.recordError(e);
@@ -112,13 +112,16 @@ private void processInternal(Exchange exchange) throws Exception {
             observation.close();
         }
 
+        // this operation is used to gate untrusted content, so a missing verdict must fail the exchange
+        // instead of leaving CamelOpenAIModerationFlagged false and letting the message through
         if (response.results().size() != inputs.size()) {
             throw new CamelExchangeException(
                     "Moderation returned " + response.results().size() + " result(s) for " + inputs.size()
                                              + " input(s)",
                     exchange);
         }
 
+        // stored only once the response is known to be complete, so a failed exchange carries no verdict
         if (config.isStoreFullResponse()) {
             exchange.setProperty(OpenAIConstants.MODERATION_RESPONSE, response);
         }

components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIResponsesProducer.java

@@ -20,7 +20,6 @@
 
 import com.openai.models.responses.Response;
 import com.openai.models.responses.ResponseCreateParams;
-import com.openai.models.responses.StructuredResponse;
 import com.openai.models.responses.StructuredResponseCreateParams;
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.Exchange;
@@ -34,6 +33,7 @@
 import org.apache.camel.support.DefaultAsyncProducer;
 import org.apache.camel.support.ResourceHelper;
 import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.function.ThrowingSupplier;
 
 /**
  * OpenAI producer for the Responses API (non-streaming).
@@ -164,29 +164,18 @@ private void processStructured(
     }
 
     private Response createResponse(Exchange exchange, String model, ResponseCreateParams params) throws Exception {
-        GenAiObservationContext observationContext = GenAiObservationContext.builder()
-                .operationName(GenAiOperationName.CHAT)
-                .system("openai")
-                .requestModel(model)
-                .componentScheme("openai")
-                .build();
-        GenAiObservation observation = GenAiObservability.start(exchange, observationContext);
-        try {
-            Response response = getEndpoint().getClient().responses().create(params);
-            recordResponseSuccess(observation, response);
-            return response;
-        } catch (Exception e) {
-            GenAiErrorSupport.apply(exchange, e);
-            observation.recordError(e);
-            throw e;
-        } finally {
-            observation.close();
-        }
+        return observedCall(exchange, model, () -> getEndpoint().getClient().responses().create(params));
     }
 
     private Response createStructuredResponse(
             Exchange exchange, String model, StructuredResponseCreateParams<?> structuredParams)
             throws Exception {
+        return observedCall(exchange, model,
+                () -> getEndpoint().getClient().responses().create(structuredParams).rawResponse());
+    }
+
+    private Response observedCall(Exchange exchange, String model, ThrowingSupplier<Response, Exception> call)
+            throws Exception {
         GenAiObservationContext observationContext = GenAiObservationContext.builder()
                 .operationName(GenAiOperationName.CHAT)
                 .system("openai")
@@ -195,10 +184,9 @@ private Response createStructuredResponse(
                 .build();
         GenAiObservation observation = GenAiObservability.start(exchange, observationContext);
         try {
-            StructuredResponse<?> structured = getEndpoint().getClient().responses().create(structuredParams);
-            Response raw = structured.rawResponse();
-            recordResponseSuccess(observation, raw);
-            return raw;
+            Response response = call.get();
+            recordResponseSuccess(observation, response);
+            return response;
         } catch (Exception e) {
             GenAiErrorSupport.apply(exchange, e);
             observation.recordError(e);
@@ -214,15 +202,11 @@ private static void recordResponseSuccess(GenAiObservation observation, Response
                 .orElse(null);
         response.usage().ifPresentOrElse(
                 usage -> observation.recordSuccess(GenAiUsage.of(
-                        toTokenCount(usage.inputTokens()),
-                        toTokenCount(usage.outputTokens()),
+                        usage.inputTokens(),
+                        usage.outputTokens(),
                         finishReason,
                         response.model().toString())),
-                () -> observation.recordSuccess(GenAiUsage.of(null, null, finishReason, response.model().toString())));
-    }
-
-    private static Integer toTokenCount(long tokens) {
-        return Math.toIntExact(tokens);
+                () -> observation.recordSuccess(GenAiUsage.of((Long) null, null, finishReason, response.model().toString())));
     }
 
     private void finishExchange(Exchange exchange, OpenAIConfiguration config, Response response, String body) {