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

Two coupled fixes to online scoring. They ship together because the
second only becomes a correctness problem once the first lands.

Retry classification. scoreTrace answered every provider failure with a
blanket InternalServerErrorException, so a request the provider can
never accept -- an oversized body rejected with a plain-text 400 -- was
replayed maxRetries times, one delivery per pendingMessageDuration,
before being retired. It now classifies by status code rather than by
4xx/5xx family: 400/401/403 and friends become ClientErrorException,
which BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS retires on the first
delivery, while 408/425/429 and every 5xx stay a retryable 500. Family
alone is too blunt -- 408/425/429 are "not now", not "not ever".

Permanence is decided only from a status the provider actually put on
the wire. The provider mappers are not consulted, not even as a
fallback: they synthesize a code when they cannot parse the body
(CustomLlm 400, OpenAi 500), and nothing downstream can tell a parsed
400 from that default, so trusting them would drop every unparseable
CustomLlm failure on its first delivery. Needlessly retrying a
permanent error costs maxRetries attempts; dropping an unknown failure
loses it forever, so an absent wire status stays retryable.

VertexAI was the one provider this could not reach. The Google Cloud
SDK raises GAX ApiException, which carries no HttpException and which
langchain4j does not model as NonRetriableException, so a permanent
Vertex failure was retried in-process and then redelivered by the
subscriber. Its status now comes from GAX's own transport-neutral
translation, identical for the gRPC and HTTP-JSON transports, and a
permanent status additionally fails fast inside the retry action so the
in-process attempts are skipped too. An exception GAX itself marks
retryable yields no status, so the mapping can only ever prevent a
drop, never cause one.

Thread fan-out. The two trace-thread scorers took a message carrying a
list of thread ids, fanned out, and collapsed all per-thread errors
into one arbitrary re-emitted error. That was harmless only while every
provider failure was retryable; after the split a fan-out can mix a
permanent 400 with a transient 429, and an arbitrary pick either drops
retryable work or replays threads that already succeeded.

Rather than choose a better victim, remove the collapse: the subscriber
acks and removes per stream entry, so OnlineScorePublisher now writes
one entry per thread id and failure granularity matches ack
granularity. A retry re-runs only the thread that failed, so no sibling
is ever replayed.

Entries left by the previous build carry several ids. Those are
migrated, not scored: the consumer republishes them as N single-id
entries and completes, which is what acks the original. A failed
republish does not ack, so the entry redelivers and the split retries;
a successful republish whose ack fails splits twice, and the duplicate
scores are absorbed by feedback_scores being a ReplacingMergeTree. The
branch is a shim, deletable once no pre-deploy entry can be in flight.
The pipeline lives in OnlineScoringBaseScorer so it cannot drift
between the two scorers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: thiagohora

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

@@ -3,6 +3,7 @@
 import com.comet.opik.api.FeedbackScoreItem;
 import com.comet.opik.api.Span;
 import com.comet.opik.api.Trace;
+import com.comet.opik.api.Visibility;
 import com.comet.opik.api.evaluators.AutomationRuleEvaluatorType;
 import com.comet.opik.api.events.RedisSubscriberMessage;
 import com.comet.opik.api.filter.Operator;
@@ -12,6 +13,7 @@
 import com.comet.opik.domain.SpanService;
 import com.comet.opik.domain.TraceSearchCriteria;
 import com.comet.opik.domain.TraceService;
+import com.comet.opik.domain.evaluators.OnlineScorePublisher;
 import com.comet.opik.infrastructure.OnlineScoringConfig;
 import com.comet.opik.infrastructure.OnlineScoringStreamConfigurationAdapter;
 import com.comet.opik.infrastructure.auth.RequestContext;
@@ -31,6 +33,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static com.comet.opik.api.FeedbackScoreItem.FeedbackScoreBatchItem;
@@ -252,4 +255,44 @@ protected Flux<Trace> retrieveFullThreadContext(@NotNull String threadId,
                                     .defer(() -> retrieveFullThreadContext(threadId, lastReceivedIdRef, projectId)));
                 }));
     }
+
+    /**
+     * Routes a trace-thread entry down one of two branches: an entry carrying several thread ids was
+     * written by an older build and is <b>migrated</b> — republished as one entry per id, never scored —
+     * while the single-id entries this build writes are <b>scored</b> normally. Completion therefore does
+     * not mean "scored", so per-outcome logging belongs inside {@code scoreThread} or in the migrate
+     * branch, never on the returned Mono.
+     *
+     * <p>Migrating rather than scoring avoids needing a rule for reducing N per-thread outcomes into the
+     * one verdict a stream entry gets, which must mis-serve somebody when a permanent failure and a
+     * retryable one land together. Each replacement entry gets its own retry budget instead.
+     *
+     * <p>The ack is implicit: this returns the republish, so a failure leaves the entry unacked and it
+     * redelivers. If the republish succeeds but the ack fails, the entry splits twice and some threads are
+     * scored twice — tolerable because {@code feedback_scores} is a {@code ReplacingMergeTree} versioned on
+     * {@code last_updated_at}, so the second score overwrites rather than duplicating.
+     *
+     * <p>The migrate branch is a temporary shim, deletable once no pre-split entry can be in flight.
+     */
+    protected Mono<Void> migrateOrScoreThreadIds(@NonNull M message, @NonNull List<String> threadIds,
+            @NonNull OnlineScorePublisher publisher, @NonNull Function<String, M> singleThreadIdCopy,
+            @NonNull Function<String, Mono<Void>> scoreThread) {
+        if (threadIds.size() > 1) {
+            // Logged on success, and worded so it can never be mistaken for a scoring log.
+            return publisher.enqueueMessage(threadIds.stream().map(singleThreadIdCopy).toList(), type)
+                    .doOnSuccess(unused -> log.info(
+                            "Migrated '{}' legacy thread ids to single-id entries for workspace '{}'; "
+                                    + "no scoring performed for this entry",
+                            threadIds.size(), message.workspaceId()));
+        }
+        // @NotEmpty says this cannot happen; if it does, no retry would help, so let it be acked away.
+        if (threadIds.isEmpty()) {
+            log.warn("Discarding trace-thread entry with no thread ids for workspace '{}'", message.workspaceId());
+            return Mono.empty();
+        }
+        return scoreThread.apply(threadIds.getFirst())
+                .contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
+                        .put(RequestContext.USER_NAME, message.userName())
+                        .put(RequestContext.VISIBILITY, Visibility.PRIVATE));
+    }
 }

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

@@ -4,7 +4,6 @@
 import com.comet.opik.api.ScoreSource;
 import com.comet.opik.api.Span;
 import com.comet.opik.api.Trace;
-import com.comet.opik.api.Visibility;
 import com.comet.opik.api.attachment.EntityType;
 import com.comet.opik.api.evaluators.AutomationRuleEvaluator;
 import com.comet.opik.api.evaluators.LlmAsJudgeMessage;
@@ -19,6 +18,7 @@
 import com.comet.opik.domain.evaluation.EvaluationRecorder;
 import com.comet.opik.domain.evaluation.OnlineEvaluationRecorder;
 import com.comet.opik.domain.evaluators.AutomationRuleEvaluatorService;
+import com.comet.opik.domain.evaluators.OnlineScorePublisher;
 import com.comet.opik.domain.evaluators.UserLog;
 import com.comet.opik.domain.llm.ChatCompletionService;
 import com.comet.opik.domain.llm.LlmProviderFactory;
@@ -35,7 +35,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RedissonReactiveClient;
 import org.slf4j.Logger;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 import ru.vyarus.dropwizard.guice.module.installer.feature.eager.EagerSingleton;
@@ -70,6 +69,7 @@ public class OnlineScoringTraceThreadLlmAsJudgeScorer extends OnlineScoringBaseS
     private final ServiceTogglesConfig serviceTogglesConfig;
     private final OnlineEvaluationRecorder onlineEvaluationRecorder;
     private final AttachmentService attachmentService;
+    private final OnlineScorePublisher onlineScorePublisher;
 
     @Inject
     public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring") OnlineScoringConfig config,
@@ -85,7 +85,8 @@ public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring"
             @NonNull AgenticScoringService agenticScoringService,
             @NonNull SpanService spanService,
             @NonNull OnlineEvaluationRecorder onlineEvaluationRecorder,
-            @NonNull AttachmentService attachmentService) {
+            @NonNull AttachmentService attachmentService,
+            @NonNull OnlineScorePublisher onlineScorePublisher) {
         super(config, redisson, feedbackScoreService, traceService, spanService, TRACE_THREAD_LLM_AS_JUDGE,
                 Constants.TRACE_THREAD_LLM_AS_JUDGE);
         this.aiProxyService = aiProxyService;
@@ -97,6 +98,7 @@ public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring"
         this.serviceTogglesConfig = serviceTogglesConfig;
         this.onlineEvaluationRecorder = onlineEvaluationRecorder;
         this.attachmentService = attachmentService;
+        this.onlineScorePublisher = onlineScorePublisher;
         this.userFacingLogger = UserFacingLoggingFactory.getLogger(OnlineScoringTraceThreadLlmAsJudgeScorer.class);
     }
 
@@ -112,25 +114,14 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreLlmAsJudge message) {
         log.info("Message received with projectId: '{}', ruleId: '{}', threadIds: '{}' for workspace '{}'",
                 message.projectId(), message.ruleId(), message.threadIds(), message.workspaceId());
 
-        return Flux.fromIterable(message.threadIds())
-                // Score each thread id independently: a single thread's failure must not stop scoring the
-                // sibling thread ids. Per-thread errors are materialized (onErrorResume) so the flatMap
-                // completes for every thread; the batch's first failure is then re-surfaced below. This keeps
-                // the failure on the Mono error path handled by BaseRedisSubscriber.processMessage's
-                // onErrorResume — classified as a processing error, following the normal retryable/
-                // non-retryable path — instead of leaking into the enclosing onErrorContinue via Flux.flatMap
-                // (which would drop the element and count it as an "unexpected" error).
-                .flatMap(threadId -> processThreadScores(message, threadId)
-                        .then(Mono.<Throwable>empty())
-                        .onErrorResume(Mono::just))
-                .collectList()
-                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
-                .contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
-                        .put(RequestContext.USER_NAME, message.userName())
-                        .put(RequestContext.VISIBILITY, Visibility.PRIVATE))
-                .doOnSuccess(unused -> log.info(
-                        "Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
-                        message.projectId(), message.ruleId(), message.workspaceId()))
+        // The success log sits inside the scoring callback: a migrated entry completes this chain without
+        // scoring anything, so a doOnSuccess out here would claim work that never happened.
+        return migrateOrScoreThreadIds(message, message.threadIds(), onlineScorePublisher,
+                threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
+                threadId -> processThreadScores(message, threadId)
+                        .doOnSuccess(unused -> log.info(
+                                "Processed trace thread '{}' for projectId '{}', ruleId '{}' for workspace '{}'",
+                                threadId, message.projectId(), message.ruleId(), message.workspaceId())))
                 .doOnError(error -> log.error(
                         "Error processing trace thread for projectId '{}', ruleId '{}' for workspace '{}'",
                         message.projectId(), message.ruleId(), message.workspaceId(), error))

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

@@ -4,14 +4,14 @@
 import com.comet.opik.api.ScoreSource;
 import com.comet.opik.api.Span;
 import com.comet.opik.api.Trace;
-import com.comet.opik.api.Visibility;
 import com.comet.opik.api.evaluators.AutomationRuleEvaluator;
 import com.comet.opik.api.events.TraceThreadToScoreUserDefinedMetricPython;
 import com.comet.opik.domain.FeedbackScoreService;
 import com.comet.opik.domain.ProjectService;
 import com.comet.opik.domain.SpanService;
 import com.comet.opik.domain.TraceService;
 import com.comet.opik.domain.evaluators.AutomationRuleEvaluatorService;
+import com.comet.opik.domain.evaluators.OnlineScorePublisher;
 import com.comet.opik.domain.evaluators.UserLog;
 import com.comet.opik.domain.evaluators.python.PythonEvaluatorService;
 import com.comet.opik.domain.threads.TraceThreadService;
@@ -26,7 +26,6 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.redisson.api.RedissonReactiveClient;
 import org.slf4j.Logger;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 import ru.vyarus.dropwizard.guice.module.installer.feature.eager.EagerSingleton;
@@ -62,6 +61,7 @@ public class OnlineScoringTraceThreadUserDefinedMetricPythonScorer
     private final ProjectService projectService;
     private final AutomationRuleEvaluatorService automationRuleEvaluatorService;
     private final AgenticScoringService agenticScoringService;
+    private final OnlineScorePublisher onlineScorePublisher;
 
     @Inject
     public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
@@ -75,7 +75,8 @@ public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
             @NonNull ProjectService projectService,
             @NonNull AutomationRuleEvaluatorService automationRuleEvaluatorService,
             @NonNull SpanService spanService,
-            @NonNull AgenticScoringService agenticScoringService) {
+            @NonNull AgenticScoringService agenticScoringService,
+            @NonNull OnlineScorePublisher onlineScorePublisher) {
         super(config, redisson, feedbackScoreService, traceService, spanService,
                 TRACE_THREAD_USER_DEFINED_METRIC_PYTHON,
                 Constants.TRACE_THREAD_USER_DEFINED_METRIC_PYTHON);
@@ -85,6 +86,7 @@ public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
         this.projectService = projectService;
         this.automationRuleEvaluatorService = automationRuleEvaluatorService;
         this.agenticScoringService = agenticScoringService;
+        this.onlineScorePublisher = onlineScorePublisher;
         this.userFacingLogger = UserFacingLoggingFactory
                 .getLogger(OnlineScoringTraceThreadUserDefinedMetricPythonScorer.class);
     }
@@ -104,25 +106,14 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreUserDefinedMetricPython me
         log.info("Message received with projectId '{}', ruleId '{}' for workspace '{}'",
                 message.projectId(), message.ruleId(), message.workspaceId());
 
-        return Flux.fromIterable(message.threadIds())
-                // Score each thread id independently: a single thread's failure must not stop scoring the
-                // sibling thread ids. Per-thread errors are materialized (onErrorResume) so the flatMap
-                // completes for every thread; the batch's first failure is then re-surfaced below. This keeps
-                // the failure on the Mono error path handled by BaseRedisSubscriber.processMessage's
-                // onErrorResume — classified as a processing error, following the normal retryable/
-                // non-retryable path — instead of leaking into the enclosing onErrorContinue via Flux.flatMap
-                // (which would drop the element and count it as an "unexpected" error).
-                .flatMap(threadId -> processThreadScores(message, threadId)
-                        .then(Mono.<Throwable>empty())
-                        .onErrorResume(Mono::just))
-                .collectList()
-                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
-                .contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
-                        .put(RequestContext.USER_NAME, message.userName())
-                        .put(RequestContext.VISIBILITY, Visibility.PRIVATE))
-                .doOnSuccess(unused -> log.info(
-                        "Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
-                        message.projectId(), message.ruleId(), message.workspaceId()))
+        // The success log sits inside the scoring callback: a migrated entry completes this chain without
+        // scoring anything, so a doOnSuccess out here would claim work that never happened.
+        return migrateOrScoreThreadIds(message, message.threadIds(), onlineScorePublisher,
+                threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
+                threadId -> processThreadScores(message, threadId)
+                        .doOnSuccess(unused -> log.info(
+                                "Processed trace thread '{}' for projectId '{}', ruleId '{}' for workspace '{}'",
+                                threadId, message.projectId(), message.ruleId(), message.workspaceId())))
                 .doOnError(error -> log.error(
                         "Error processing trace thread for projectId '{}', ruleId '{}' for workspace '{}'",
                         message.projectId(), message.ruleId(), message.workspaceId(), error))

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

@@ -185,8 +185,8 @@ private Mono<Integer> evaluateTraces(List<UUID> traceIds, List<AutomationRuleEva
                 .toList();
         Mono<Void> traceThreadMono = Flux.fromIterable(traceThreadRules)
                 .flatMap(rule -> {
-                    log.info("Enqueueing trace-thread evaluation for rule '{}' with '{}' trace IDs", rule.getId(),
-                            traceIdStrings.size());
+                    log.info("Enqueueing '{}' trace-thread evaluation messages, one per trace ID, for rule '{}'",
+                            traceIdStrings.size(), rule.getId());
                     return onlineScorePublisher.enqueueThreadMessage(traceIdStrings, rule, projectId, workspaceId,
                             userName);
                 })
@@ -384,8 +384,8 @@ private Mono<Integer> evaluateThreads(List<UUID> threadModelIds, List<Automation
                     // reactive context. enqueueThreadMessage does a blocking rule lookup, so defer onto boundedElastic.
                     return Flux.fromIterable(rules)
                             .flatMap(rule -> {
-                                log.info("Enqueueing evaluation for rule '{}' with '{}' thread IDs", rule.getId(),
-                                        threadIds.size());
+                                log.info("Enqueueing '{}' evaluation messages, one per thread ID, for rule '{}'",
+                                        threadIds.size(), rule.getId());
                                 return onlineScorePublisher.enqueueThreadMessage(threadIds, rule, projectId,
                                         workspaceId, userName);
                             })