[OPIK-8262] [BE] fix: classify provider errors by status and stop collapsing trace-thread fan-out

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java

@@ -6,6 +6,7 @@
 import com.comet.opik.infrastructure.metrics.ErrorMetricsResolver;
 import com.comet.opik.infrastructure.redis.UndecodablePayloadException;
 import com.comet.opik.infrastructure.redis.UndecodableStreamMessage;
+import com.comet.opik.utils.HttpStatusRetryability;
 import io.dropwizard.lifecycle.Managed;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
@@ -58,14 +59,17 @@ public abstract class BaseRedisSubscriber<M> implements Managed {
     private static final String NOGROUP = "NOGROUP";
 
     /**
-     * Non-retryable exception types that won't succeed on retry. Checked via {@code instanceof} in
-     * {@link #isRetryableException(Throwable)}, so subclasses are automatically covered.
-     * These are usually programming, validation, client etc. exceptions.
+     * Exception types that mean the code has a bug, so retrying would just rerun the bug. Checked via
+     * {@code instanceof} in {@link #isRetryableException(Throwable)}, so subclasses are covered too.
+     *
+     * <p>{@code ClientErrorException} used to be listed here and no longer is. It is a transport type, not a
+     * bug signal, and matching it by class made the whole 4xx family permanent — including 408, 425 and 429,
+     * which are transient by definition. It is now classified by the status it carries; see
+     * {@link #isRetryableException(Throwable)}.
      */
     private static final Set<Class<? extends RuntimeException>> NON_RETRYABLE_EXCEPTIONS = Set.of(
             ArithmeticException.class,
             ClassCastException.class,
-            ClientErrorException.class,
             IllegalArgumentException.class,
             IllegalStateException.class,
             IndexOutOfBoundsException.class,
@@ -828,13 +832,19 @@ private Optional<Long> extractTimeFromMessageId(StreamMessageId messageId) {
     }
 
     /**
-     * Non-retryable exceptions are checked via {@code instanceof} against {@link #NON_RETRYABLE_EXCEPTIONS},
-     * so both exact types and their subclasses are covered.
-     * Non-retryable exceptions are usually programming, validation, client errors that won't succeed on retry.
-     * All other exceptions are considered retryable (transient errors like network issues, timeouts, server errors, etc.)
-     * Unknown exceptions default to retryable for safety.
+     * Whether the entry is worth redelivering. A {@link ClientErrorException} is decided from the status it
+     * carries, so a 408/425/429 is retried and the rest of the 4xx family is retired on first delivery;
+     * everything else is matched by class against {@link #NON_RETRYABLE_EXCEPTIONS}, whose members all mean
+     * the code has a bug. Anything unrecognised is retryable, so an unknown failure is never silently lost.
+     *
+     * <p>Classifying by class alone is what forced {@code ChatCompletionService.scoreTrace} to report every
+     * provider failure as a blanket 500: a truthful 429 would have been dropped here. With the status
+     * consulted, that workaround is gone and the provider's real status is reported.
      */
-    private boolean isRetryableException(Throwable exception) {
+    private static boolean isRetryableException(Throwable exception) {
+        if (exception instanceof ClientErrorException clientError) {
+            return !HttpStatusRetryability.isPermanent(clientError.getResponse().getStatus());
+        }
         return NON_RETRYABLE_EXCEPTIONS.stream()
                 .noneMatch(nonRetryable -> nonRetryable.isInstance(exception));
     }

apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/OnlineScorePublisher.java

@@ -56,29 +56,29 @@ public interface OnlineScorePublisher {
     Mono<Void> enqueueMessage(List<?> messages, AutomationRuleEvaluatorType type);
 
     /**
-     * Enqueues a thread message for scoring based on the provided rule. The returned publisher must be subscribed
-     * for the enqueue to happen.
+     * Enqueues thread messages for scoring — <b>one stream entry per thread id</b>, not one for the batch.
+     * The returned publisher must be subscribed for the enqueue to happen.
      *
-     * @param threadIds   the IDs of the threads to score
+     * @param threadIds   the IDs of the threads to score; one message is published per element
      * @param ruleId      the ID of the rule to apply
      * @param projectId   the ID of the project
      * @param workspaceId the ID of the workspace
      * @param userName    the name of the user who initiated the scoring
-     * @return a {@link Mono} that completes once the message is enqueued
+     * @return a {@link Mono} that completes once all messages are enqueued
      */
     Mono<Void> enqueueThreadMessage(List<String> threadIds, UUID ruleId, UUID projectId, String workspaceId,
             String userName);
 
     /**
-     * Enqueues a thread message for an already-resolved rule, avoiding the blocking rule lookup that the
-     * {@code ruleId} overload performs. Prefer this when the caller already holds the {@link AutomationRuleEvaluator}.
+     * Enqueues thread messages for an already-resolved rule, avoiding the blocking rule lookup that the
+     * {@code ruleId} overload performs. Publishes <b>one stream entry per thread id</b>.
      *
-     * @param threadIds   the IDs of the threads to score
+     * @param threadIds   the IDs of the threads to score; one message is published per element
      * @param rule        the already-resolved automation rule evaluator
      * @param projectId   the ID of the project
      * @param workspaceId the ID of the workspace
      * @param userName    the name of the user who initiated the scoring
-     * @return a {@link Mono} that completes once the message is enqueued
+     * @return a {@link Mono} that completes once all messages are enqueued
      */
     Mono<Void> enqueueThreadMessage(List<String> threadIds, AutomationRuleEvaluator<?, ?> rule, UUID projectId,
             String workspaceId, String userName);
@@ -185,16 +185,27 @@ public Mono<Void> enqueueThreadMessage(@NonNull List<String> threadIds,
             @NonNull AutomationRuleEvaluator<?, ?> rule, @NonNull UUID projectId, @NonNull String workspaceId,
             @NonNull String userName) {
 
-        // Caller already holds the resolved rule — no findById needed.
+        // Caller already holds the resolved rule -- no findById needed.
+        //
+        // One message PER THREAD ID, not one carrying the whole list: the subscriber acks and removes per
+        // stream entry, so an entry holding N ids forces N independent outcomes through a single verdict.
+        // Splitting means a retry replays exactly the thread that failed. Costs N entries where there was
+        // one; the streams are capped by streamMaxLen and trimmed non-strictly.
         return switch (rule) {
             case AutomationRuleEvaluatorTraceThreadLlmAsJudge llmAsJudge -> enqueueMessage(
-                    List.of(toLlmAsJudgeMessage(threadIds, rule.getId(), projectId, workspaceId, userName,
-                            llmAsJudge.getCode())),
+                    threadIds.stream()
+                            .map(threadId -> toLlmAsJudgeMessage(threadId, rule.getId(), projectId, workspaceId,
+                                    userName, llmAsJudge.getCode()))
+                            .toList(),
                     rule.getType());
             case AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython definedMetricPython -> {
                 if (serviceTogglesConfig.isTraceThreadPythonEvaluatorEnabled()) {
-                    yield enqueueMessage(List.of(toDefinedMetricPython(threadIds, rule.getId(), projectId,
-                            workspaceId, userName, definedMetricPython.getCode())), rule.getType());
+                    yield enqueueMessage(
+                            threadIds.stream()
+                                    .map(threadId -> toDefinedMetricPython(threadId, rule.getId(), projectId,
+                                            workspaceId, userName, definedMetricPython.getCode()))
+                                    .toList(),
+                            rule.getType());
                 }
                 log.warn("Trace Thread online scoring python evaluator is disabled, skipping enqueueing "
                         + "for ruleId: '{}'", rule.getId());
@@ -204,10 +215,14 @@ yield enqueueMessage(List.of(toDefinedMetricPython(threadIds, rule.getId(), proj
         };
     }
 
-    private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(List<String> threadIds, UUID ruleId, UUID projectId,
+    /**
+     * {@code threadIds} stays a list even though this puts one id in it: narrowing the field would make
+     * the multi-id entries left by the previous build undecodable during the rolling upgrade.
+     */
+    private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(String threadId, UUID ruleId, UUID projectId,
             String workspaceId, String userName, TraceThreadLlmAsJudgeCode code) {
         return TraceThreadToScoreLlmAsJudge.builder()
-                .threadIds(threadIds)
+                .threadIds(List.of(threadId))
                 .ruleId(ruleId)
                 .projectId(projectId)
                 .workspaceId(workspaceId)
@@ -216,10 +231,11 @@ private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(List<String> threadIds,
                 .build();
     }
 
-    private TraceThreadToScoreUserDefinedMetricPython toDefinedMetricPython(List<String> threadIds, UUID ruleId,
+    /** @see #toLlmAsJudgeMessage on why {@code threadIds} stays a list. */
+    private TraceThreadToScoreUserDefinedMetricPython toDefinedMetricPython(String threadId, UUID ruleId,
             UUID projectId, String workspaceId, String userName, TraceThreadUserDefinedMetricPythonCode code) {
         return TraceThreadToScoreUserDefinedMetricPython.builder()
-                .threadIds(threadIds)
+                .threadIds(List.of(threadId))
                 .ruleId(ruleId)
                 .projectId(projectId)
                 .workspaceId(workspaceId)

apps/opik-backend/src/main/java/com/comet/opik/domain/llm/ChatCompletionService.java

@@ -3,7 +3,11 @@
 import com.comet.opik.api.evaluators.LlmAsJudgeModelParameters;
 import com.comet.opik.infrastructure.LlmProviderClientConfig;
 import com.comet.opik.utils.ChunkedOutputHandlers;
+import com.comet.opik.utils.HttpStatusRetryability;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.common.base.Throwables;
+import com.openai.errors.OpenAIServiceException;
 import dev.langchain4j.exception.AuthenticationException;
 import dev.langchain4j.exception.HttpException;
 import dev.langchain4j.exception.InternalServerException;
@@ -135,29 +139,37 @@ public ChatResponse scoreTrace(@NonNull ChatRequest chatRequest,
         try {
             log.info("Initiating chat with model '{}' expecting structured response, workspaceId '{}'",
                     modelParameters.name(), workspaceId);
-            chatResponse = retryPolicy
-                    .withRetry(() -> failFastOnUnsupportedFeature(() -> languageModelClient.chat(chatRequest)));
+            chatResponse = retryPolicy.withRetry(
+                    () -> failFastOnPermanentFailure(
+                            () -> failFastOnUnsupportedFeature(() -> languageModelClient.chat(chatRequest))));
             log.info("Completed chat with model '{}' expecting structured response, workspaceId '{}'",
                     modelParameters.name(), workspaceId);
             return chatResponse;
         } catch (RuntimeException runtimeException) {
             failIfUnsupportedFeature(runtimeException);
 
-            LlmProviderService provider = llmProviderFactory.getService(workspaceId, modelParameters.name());
+            // Report the status the provider actually sent, same as create() and the streaming handler.
+            // BaseRedisSubscriber classifies a ClientErrorException by that status, so a truthful 429 is
+            // redelivered and a truthful 400 is retired; this no longer has to misreport either.
+            //
+            // Only a wire status is used. The provider mappers synthesize one when they cannot parse the body
+            // (CustomLlm 400, OpenAi 500) and nothing downstream can tell that from a real 400, so consulting
+            // them would retire every unparseable CustomLlm failure on its first delivery. An absent status
+            // falls through to 500 and stays retryable: burning maxRetries on a doomed request costs attempts,
+            // losing an unknown failure costs the evaluation.
+            var status = findProviderHttpStatus(runtimeException)
+                    .orElse(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
 
-            Optional<ErrorMessage> providerError = provider.getLlmProviderError(runtimeException);
-
-            providerError
-                    .ifPresent(llmProviderError -> failHandlingLLMProviderError(runtimeException, llmProviderError));
-
-            // No failIfProviderReportedHttpStatus here, unlike create() and the streaming handler. This method is
-            // called only by the online-scoring subscribers, never from a resource, so a recovered status reaches no
-            // HTTP client — while BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS lists ClientErrorException, so turning
-            // a rate limit into a 429 or a provider timeout into a 408 would make the subscriber ack and drop the
-            // evaluation instead of honouring onlineScoring.maxRetries. Both are RetriableException upstream, so the
-            // blanket 500 is what keeps them retryable.
             log.warn(UNEXPECTED_ERROR_CALLING_LLM_PROVIDER, runtimeException);
-            throw new InternalServerErrorException(buildDetailedErrorMessage(runtimeException), runtimeException);
+
+            var detail = buildDetailedErrorMessage(runtimeException);
+            // findProviderHttpStatus only yields error statuses, so these two families are exhaustive. The
+            // cause is carried through, unlike failHandlingLLMProviderError, because this exception is logged
+            // rather than serialized to an HTTP caller and the provider stack is the diagnostic.
+            if (familyOf(status) == Response.Status.Family.CLIENT_ERROR) {
+                throw new ClientErrorException(detail, status, runtimeException);
+            }
+            throw new ServerErrorException(detail, status, runtimeException);
         } finally {
             // Close the Vertex client (reused across retries) to release its GAX threads; other providers self-reclaim.
             if (languageModelClient instanceof AutoCloseable closeable) {
@@ -188,6 +200,25 @@ private <T> T failFastOnUnsupportedFeature(Callable<T> action) throws Exception
         }
     }
 
+    /**
+     * Skips the in-process retries for a provider status that can never succeed, mirroring
+     * {@link #failFastOnUnsupportedFeature}. Applied only on the scoreTrace path: {@code create()} answers an HTTP
+     * caller, and narrowing its retry behaviour is not this change's business. Mainly reached for VertexAI, whose GAX
+     * exceptions langchain4j does not model as {@code NonRetriableException}; the mapped providers already fail fast
+     * on their own. The cause is preserved, so the catch block still classifies from the same status.
+     */
+    private <T> T failFastOnPermanentFailure(Callable<T> action) throws Exception {
+        try {
+            return action.call();
+        } catch (RuntimeException runtimeException) {
+            if (findProviderHttpStatus(runtimeException).filter(HttpStatusRetryability::isPermanent)
+                    .isPresent()) {
+                throw new NonRetriableException(runtimeException);
+            }
+            throw runtimeException;
+        }
+    }
+
     /**
      * langchain4j raises {@link UnsupportedFeatureException} when the request asks for a capability the selected
      * provider does not implement — e.g. {@code ToolChoice.REQUIRED} against Vertex AI Gemini. The provider is never
@@ -292,6 +323,22 @@ private static boolean isErrorStatus(int status) {
         return family == Response.Status.Family.CLIENT_ERROR || family == Response.Status.Family.SERVER_ERROR;
     }
 
+    /**
+     * VertexAI is one of two providers whose client raises no {@link HttpException}: the Google Cloud SDK throws GAX
+     * {@code ApiException}, which is also not a {@code NonRetriableException}, so without this a permanent Vertex
+     * failure consumed the whole retry budget. The status is GAX's own transport-neutral translation, identical for
+     * the gRPC and HTTP-JSON transports, rather than a table of our own. An exception GAX itself marks retryable
+     * yields no status at all, so this can only ever prevent a drop, never cause one.
+     */
+    private static Optional<Integer> gaxHttpStatus(ApiException apiException) {
+        if (apiException.isRetryable()) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(apiException.getStatusCode())
+                .map(StatusCode::getCode)
+                .map(StatusCode.Code::getHttpStatusCode);
+    }
+
     /**
      * The status langchain4j's own exception types stand for, used for providers whose clients raise them without an
      * {@link HttpException} in the chain. {@code ContentFilteredException} is covered by its
@@ -305,6 +352,8 @@ private Optional<Integer> canonicalStatusOf(Throwable throwable) {
             case TimeoutException ignored -> Optional.of(Response.Status.REQUEST_TIMEOUT.getStatusCode());
             case RateLimitException ignored -> Optional.of(Response.Status.TOO_MANY_REQUESTS.getStatusCode());
             case InternalServerException ignored -> Optional.of(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
+            case ApiException apiException -> gaxHttpStatus(apiException);
+            case OpenAIServiceException responsesException -> Optional.of(responsesException.statusCode());
             default -> Optional.empty();
         };
     }