[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;
@@ -229,6 +232,76 @@ private static <T extends FeedbackScoreItem> Map<String, List<BigDecimal>> group
      * @return a Flux of Trace objects representing the full thread context
      */
     //TODO: Move this to a common service or utility class
+    /**
+     * The whole per-entry pipeline for a trace-thread scorer: migrate a legacy multi-id entry, or score
+     * the single thread id this build's entries carry. Shared by both trace-thread scorers so the
+     * migration path cannot drift between them — message construction and the scorer's own logging stay
+     * with the caller, only the branching lives here.
+     *
+     * <p><b>One thread id per entry.</b> Since OPIK-8262 the publisher writes one entry per thread id, so
+     * failure granularity matches {@link BaseRedisSubscriber}'s per-entry ack/remove granularity: a retry
+     * re-runs only the thread that failed, and no sibling is ever replayed. The single-id branch therefore
+     * just returns the scoring Mono — its error reaches
+     * {@code BaseRedisSubscriber.processMessage}'s {@code onErrorResume} directly and is classified
+     * retryable or not for this entry alone. There is deliberately no {@code Flux.flatMap} here, so
+     * nothing can leak into the enclosing {@code onErrorContinue} and be miscounted as an "unexpected"
+     * error.
+     *
+     * <p><b>Multi-id entries are migrated, not scored (rolling-upgrade shim).</b> Entries written by the
+     * previous build carry several ids. Scoring those in place would need a rule for reducing N per-thread
+     * outcomes into the one verdict the entry gets — a second set of semantics, and one that must
+     * mis-serve somebody whenever a permanent failure and a retryable one land under the same entry.
+     * Converting the entry to the new format instead means there is no verdict to pick between siblings,
+     * no amplification, and nothing lost: each thread id becomes its own entry with its own retry budget,
+     * scored by the ordinary single-id path.
+     *
+     * <p><b>Ack is implicit, and that is what makes the failure modes right.</b> The republish itself is
+     * returned, so the base subscriber's normal success/failure handling supplies the ack:
+     * <ul>
+     *   <li><b>Republish fails</b> — the returned Mono errors, the entry is <em>not</em> acked, and it
+     *       redelivers so the split is retried. Nothing is lost.</li>
+     *   <li><b>Republish succeeds, ack/remove fails</b> — the entry redelivers and splits again, so some
+     *       thread ids get duplicate single-id entries and are scored twice. Tolerable rather than merely
+     *       unlikely: {@code feedback_scores} is a {@code ReplicatedReplacingMergeTree} versioned on
+     *       {@code last_updated_at} (migration {@code 000017}), so the second score overwrites the first
+     *       rather than accumulating a duplicate row. The cost is wasted provider calls, not wrong data.</li>
+     * </ul>
+     *
+     * <p><b>The multi-id branch is temporary.</b> It can only fire for entries written before the deploy
+     * that ships OPIK-8262. Once no such entry can be in flight — one full {@code streamMaxLen} turnover
+     * past the rollout, at the latest — this branch, and the {@code List<String>} shape of
+     * {@code threadIds}, are both deletable.
+     *
+     * @param message             the entry being processed, read for its workspace and user
+     * @param threadIds           the entry's thread ids; exactly one for anything this build wrote
+     * @param publisher           writes the replacement entries onto this scorer's own stream
+     * @param singleThreadIdCopy  builds a copy of {@code message} carrying only the given thread id; the
+     *                            rule code, workspace and user are preserved verbatim, so no rule lookup
+     *                            or evaluator-toggle re-check is involved in the migration
+     * @param scoreThread         scores one thread id, on the ordinary path
+     * @return a {@link Mono} completing when the entry has been scored or migrated
+     */
+    protected Mono<Void> scoreOneThreadIdPerEntry(@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) {
+            log.info("Splitting legacy entry of '{}' thread ids into one entry each for workspace '{}'",
+                    threadIds.size(), message.workspaceId());
+            return publisher.enqueueMessage(threadIds.stream().map(singleThreadIdCopy).toList(), type);
+        }
+        // @NotEmpty on the message record says this cannot happen. If it somehow does there is nothing to
+        // score and no retry could change that, so complete normally and let the subscriber ack it away
+        // rather than cycle it through maxRetries.
+        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));
+    }
+
     protected Flux<Trace> retrieveFullThreadContext(@NotNull String threadId,
             @NotNull AtomicReference<UUID> lastReceivedIdRef, @NotNull UUID projectId) {
 

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,22 +114,12 @@ 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))
+        // Branching (legacy multi-id migration vs the ordinary single-id path) lives in
+        // OnlineScoringBaseScorer so it cannot drift between the two trace-thread scorers; message
+        // construction and the logging below stay here.
+        return scoreOneThreadIdPerEntry(message, message.threadIds(), onlineScorePublisher,
+                threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
+                threadId -> processThreadScores(message, threadId))
                 .doOnSuccess(unused -> log.info(
                         "Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
                         message.projectId(), message.ruleId(), message.workspaceId()))

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,22 +106,12 @@ 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))
+        // Branching (legacy multi-id migration vs the ordinary single-id path) lives in
+        // OnlineScoringBaseScorer so it cannot drift between the two trace-thread scorers; message
+        // construction and the logging below stay here.
+        return scoreOneThreadIdPerEntry(message, message.threadIds(), onlineScorePublisher,
+                threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
+                threadId -> processThreadScores(message, threadId))
                 .doOnSuccess(unused -> log.info(
                         "Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
                         message.projectId(), message.ruleId(), message.workspaceId()))

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

@@ -185,8 +185,9 @@ 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());
+                    // One stream entry per trace ID (OPIK-8262), not one entry for the batch.
+                    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 +385,9 @@ 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());
+                                // One stream entry per thread ID (OPIK-8262), not one entry for the batch.
+                                log.info("Enqueueing '{}' evaluation messages, one per thread ID, for rule '{}'",
+                                        threadIds.size(), rule.getId());
                                 return onlineScorePublisher.enqueueThreadMessage(threadIds, rule, projectId,
                                         workspaceId, userName);
                             })