[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.

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);
                             })