[OPIK-8240] [BE] fix: split provider-error retryability by status, resume XAUTOCLAIM from its cursor

Two independent defects that each leave a permanently failing online-scoring
message cycling instead of retiring it.

1. A permanent provider 4xx was retried as if transient.

ChatCompletionService.scoreTrace answered every unmappable provider failure
with a blanket InternalServerErrorException, which sits outside
BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS -- so a request that can never
succeed was replayed maxRetries times, once per pendingMessageDuration. The
real status was available the whole time on the HttpException in the cause
chain; it went unread because the existing recovery path classifies by family,
and mapping all of 4xx to ClientErrorException would drop transient 408/429
after one attempt.

Classify by status code instead: 400/401/403 and the rest of 4xx are
permanent, 408/425/429 and all 5xx stay retryable. scoreTrace gets its own
mapping so create() and the streaming handler keep returning the provider's
status verbatim to HTTP callers. This also closes a pre-existing over-eager
drop on the parseable branch, where a JSON-parseable 429 already became
ClientErrorException and was dropped after one attempt.

2. XAUTOCLAIM never scanned past the first ~100 pending entries.

claimPendingMessages passed StreamMessageId.MIN as the scan start every call
and discarded the cursor Redis returns. Redis caps each XAUTOCLAIM at
COUNT * 10 PEL entries *examined*, so at consumerBatchSize=10 a call inspects
only the first 100 -- and restarting at MIN means nothing past that window is
ever examined. Any backlog above ~100 grows a permanently unreachable tail.

Carry getNextId() forward, resetting to MIN on Redis's 0-0 end-of-pass reply.
A failed scan deliberately does not advance the cursor, so its window is
retried rather than skipped. The sentinel is compared numerically:
StreamMessageId.MIN/.ALL serialize to "-" and "0" and neither is equals() to
the StreamMessageId(0, 0) Redisson parses 0-0 into (verified against redisson
4.7.0), so matching on the constants would never fire.

Addressed from review:

- Status precedence was backwards. The provider mappers synthesize a status
  when they cannot read one off the body (CustomLlmErrorMessage defaults to
  400, OpenAiErrorMessage to 500), and preferring the mapped code let a
  synthetic 400 mask a real upstream 503 -- classifying a transient failure as
  permanent and dropping it on first delivery. The cause-chain HttpException
  now wins, with the mapped code as fallback, matching the precedence
  findProviderHttpStatus already documents for its own chain walk.

- Sibling aggregation could discard retryable work. The trace-thread scorers
  re-emit one error for a message that fans out over many thread ids, and
  errors.getFirst() made that an arrival-order race. Harmless while every
  failure was a blanket 500; not once permanent and transient were split, as a
  ClientErrorException arriving first would ack and remove the entry with its
  retryable siblings. Both scorers now share representativeError(), which
  prefers a retryable sibling: a bounded replay is recoverable, silently
  dropped work is not.

- Test-only: unconditional assertion flow in the parameterized status test,
  real-mapper-path coverage for the precedence rule, display-name wording.

Tests: 145 green across the affected suites. Mutation-checked -- collapsing
the status split to the whole 4xx family fails 5, removing it fails 7,
reverting the cursor to always-MIN fails 3, inverting the status precedence
fails 4, reverting aggregation to getFirst() fails 1.

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/BaseRedisSubscriber.java

@@ -127,6 +127,27 @@ public abstract class BaseRedisSubscriber<M> implements Managed {
     private final LongHistogram listPendingTime;
     private final LongCounter unexpectedErrors;
 
+    /**
+     * Resume point for the next {@code XAUTOCLAIM} scan.
+     *
+     * <p>Redis caps each {@code XAUTOCLAIM} at {@code COUNT * 10} PEL entries <em>examined</em> (not
+     * claimed), so with the default {@code consumerBatchSize} of 10 a single call only ever inspects the
+     * first 100 entries of the pending list. Restarting every scan at {@link StreamMessageId#MIN} —
+     * which this class used to do, discarding the cursor Redis hands back — means anything past that
+     * first window is never examined at all: a backlog above ~100 entries develops a permanently
+     * unreachable tail, whatever the retry or decode behaviour further down. Carrying the returned
+     * cursor forward walks the whole PEL instead, a window at a time.
+     *
+     * <p>Redis returns {@code 0-0} once a pass has covered the entire pending list; that resets this
+     * back to {@link StreamMessageId#MIN} so the next pass starts over from the oldest entry, keeping
+     * the oldest-first bias the original code intended.
+     *
+     * <p>Volatile rather than synchronized: {@code concatMap} in {@link #setupStreamListener()} already
+     * serializes claim calls, so this is only ever written by one claim at a time; volatile just
+     * publishes that write to whichever scheduler thread runs the next one.
+     */
+    private volatile StreamMessageId claimCursor = StreamMessageId.MIN;
+
     private volatile RStreamReactive<String, M> stream;
     private volatile Disposable streamSubscription;
     private volatile Scheduler timerScheduler;
@@ -398,11 +419,11 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 consumerId,
                 config.getPendingMessageDuration().toJavaDuration().toMillis(),
                 TimeUnit.MILLISECONDS,
-                StreamMessageId.MIN, // Start from the beginning of pending list
+                claimCursor, // Resume where the previous scan stopped; see claimCursor
                 config.getConsumerBatchSize())
                 .subscribeOn(consumerScheduler)
                 .filter(Objects::nonNull)
-                .map(AutoClaimResult::getMessages)
+                .map(this::advanceCursorAndExtractMessages)
                 .filter(Objects::nonNull)
                 .doOnSuccess(claimedMessages -> {
                     claimedMessages = Objects.requireNonNullElse(claimedMessages, Map.of());
@@ -412,14 +433,48 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 .onErrorResume(throwable -> {
                     claimErrors.add(1);
                     log.error("Error claiming pending messages", throwable);
+                    // A failed scan leaves the cursor where it was, so the next attempt retries the same
+                    // window rather than skipping it. Except on NOGROUP: the group is being recreated, so
+                    // any cursor into the old group's PEL is meaningless.
                     if (isNoGroupError(throwable)) {
+                        claimCursor = StreamMessageId.MIN;
                         return recoverFromNoGroup();
                     }
                     return Mono.just(Map.of());
                 })
                 .doFinally(signalType -> claimTime.record(System.currentTimeMillis() - startMillis));
     }
 
+    /**
+     * Advances {@link #claimCursor} to where Redis says the next scan should resume, then returns the
+     * batch this scan actually claimed.
+     *
+     * <p>A {@code null} or {@code 0-0} next-id means the pass reached the end of the pending list, so the
+     * cursor goes back to {@link StreamMessageId#MIN} and the following pass starts from the oldest entry
+     * again.
+     */
+    private Map<StreamMessageId, Map<String, M>> advanceCursorAndExtractMessages(AutoClaimResult<String, M> result) {
+        claimCursor = nextCursor(result.getNextId());
+        return result.getMessages();
+    }
+
+    /**
+     * The cursor to use for the next scan, given the next-id Redis returned.
+     *
+     * <p>Compared numerically rather than against {@link StreamMessageId#MIN} / {@link StreamMessageId#ALL}:
+     * those are wire sentinels that serialize to {@code -} and {@code 0}, and neither is
+     * {@code equals()} to the {@code StreamMessageId(0, 0)} that Redisson parses Redis's literal
+     * {@code 0-0} end-of-pass reply into. Matching on the constants would therefore never fire, and the
+     * scan would run off the end of the PEL and stay there instead of wrapping.
+     */
+    // Package-private for unit tests.
+    static StreamMessageId nextCursor(StreamMessageId nextId) {
+        if (nextId == null || (nextId.getId0() == 0 && nextId.getId1() == 0)) {
+            return StreamMessageId.MIN;
+        }
+        return nextId;
+    }
+
     private Mono<Map<StreamMessageId, Map<String, M>>> readMessages() {
         var startMillis = System.currentTimeMillis();
         var streamReadGroupArgs = StreamReadGroupArgs.neverDelivered()

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

@@ -16,6 +16,7 @@
 import com.comet.opik.infrastructure.OnlineScoringStreamConfigurationAdapter;
 import com.comet.opik.infrastructure.auth.RequestContext;
 import jakarta.validation.constraints.NotNull;
+import jakarta.ws.rs.ClientErrorException;
 import lombok.NonNull;
 import org.apache.commons.lang3.StringUtils;
 import org.redisson.api.RedissonReactiveClient;
@@ -154,6 +155,34 @@ protected List<Span> getSpansFromPreloadAndLogOverflow(@NonNull ThreadSpanPreloa
      * scoring chain (feedback-score persistence reads it). Per-message throughput and error metrics are
      * attributed automatically by {@link BaseRedisSubscriber} from {@link #messageContext(Object)}.
      */
+    /**
+     * Picks which of several sibling failures represents the batch, when one message fans out over many
+     * thread ids and more than one of them fails.
+     *
+     * <p><b>Retryable wins.</b> The whole batch travels as a single stream entry, so the type re-emitted
+     * here decides the fate of every sibling in it: a {@code ClientErrorException} tells
+     * {@code BaseRedisSubscriber} to ack and remove, taking any retryable sibling down with it, unretried.
+     * Picking arbitrarily -- {@code errors.getFirst()}, i.e. whichever the {@code flatMap} happened to
+     * emit first -- made that a race. It was harmless while every provider failure was a blanket 500 and
+     * the choice could not change retryability; it stopped being harmless once OPIK-8240 split permanent
+     * from transient.
+     *
+     * <p>The asymmetry is deliberate. Preferring retryable costs a bounded replay of the permanent sibling
+     * ({@code maxRetries} caps it) and the permanent one is dropped for good on the final attempt.
+     * Preferring non-retryable costs recoverable work, silently and permanently. Only one of those is
+     * recoverable, so the tie goes to retrying.
+     *
+     * @param errors the sibling failures, never empty
+     * @return the first retryable failure if any, otherwise the first failure
+     */
+    // Package-private for unit tests.
+    static Throwable representativeError(@NonNull List<Throwable> errors) {
+        return errors.stream()
+                .filter(error -> !(error instanceof ClientErrorException))
+                .findFirst()
+                .orElseGet(errors::getFirst);
+    }
+
     @Override
     protected final Mono<Void> processEvent(M message) {
         var workspaceName = StringUtils.defaultIfBlank(message.workspaceName(), message.workspaceId());

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

@@ -124,7 +124,7 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreLlmAsJudge message) {
                         .then(Mono.<Throwable>empty())
                         .onErrorResume(Mono::just))
                 .collectList()
-                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
+                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(representativeError(errors)))
                 .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/OnlineScoringTraceThreadUserDefinedMetricPythonScorer.java

@@ -116,7 +116,7 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreUserDefinedMetricPython me
                         .then(Mono.<Throwable>empty())
                         .onErrorResume(Mono::just))
                 .collectList()
-                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
+                .flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(representativeError(errors)))
                 .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/domain/llm/ChatCompletionService.java

@@ -36,6 +36,7 @@
 import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.function.Consumer;
 
@@ -147,16 +148,38 @@ public ChatResponse scoreTrace(@NonNull ChatRequest chatRequest,
 
             Optional<ErrorMessage> providerError = provider.getLlmProviderError(runtimeException);
 
-            providerError
-                    .ifPresent(llmProviderError -> failHandlingLLMProviderError(runtimeException, llmProviderError));
+            // Deliberately NOT failHandlingLLMProviderError here, unlike create() and the streaming handler.
+            // Those two answer an HTTP caller, so they want the provider's status verbatim - a 429 upstream
+            // should be a 429 downstream. This method answers the online-scoring subscribers instead, where
+            // the thrown type decides retryability: BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS matches on
+            // ClientErrorException, so the whole 4xx family would be dropped on the first failure - taking
+            // genuinely transient 408s and 429s with it. Route through the retryability split instead, which
+            // separates "this request can never succeed" from "try again later" inside that family.
+            // Cause-chain HttpException FIRST, provider mapping only as fallback. The mappers synthesize
+            // a status when they cannot read one off the body -- CustomLlmErrorMessage defaults to 400,
+            // OpenAiErrorMessage to 500 -- and a synthetic value must never outrank the code the provider
+            // actually put on the wire. Getting this backwards is not cosmetic: a real upstream 503 behind
+            // an unparseable body would surface as CustomLlm's synthetic 400, be classified permanent, and
+            // have a retryable failure dropped on its first delivery. Same precedence, and same reasoning,
+            // as findProviderHttpStatus applies within its own chain walk.
+            int status = findProviderHttpStatus(runtimeException)
+                    .or(() -> providerError.map(ErrorMessage::getCode).filter(ChatCompletionService::isErrorStatus))
+                    .orElse(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
 
-            // 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);
+
+            if (isPermanentFailure(status)) {
+                // Non-retryable: the subscriber acks and removes on the first delivery instead of burning
+                // maxRetries x pendingMessageDuration on a request whose outcome cannot change. This is the
+                // OPIK-8240 case - a provider rejecting an oversized/invalid request rejects every retry of
+                // it identically.
+                throw new ClientErrorException(buildDetailedErrorMessage(runtimeException), status,
+                        runtimeException);
+            }
+            // Everything else stays a blanket 500 so it lands outside NON_RETRYABLE_EXCEPTIONS and honours
+            // onlineScoring.maxRetries. Note this keeps transient 4xx (408/429) retryable even though their
+            // real status is a client error, which is exactly the distinction ClientErrorException cannot
+            // express on its own.
             throw new InternalServerErrorException(buildDetailedErrorMessage(runtimeException), runtimeException);
         } finally {
             // Close the Vertex client (reused across retries) to release its GAX threads; other providers self-reclaim.
@@ -292,6 +315,35 @@ private static boolean isErrorStatus(int status) {
         return family == Response.Status.Family.CLIENT_ERROR || family == Response.Status.Family.SERVER_ERROR;
     }
 
+    /**
+     * Statuses in the 4xx family that a retry could still clear, so they must NOT be treated as permanent.
+     *
+     * <p>{@code 408 Request Timeout}, {@code 425 Too Early} and {@code 429 Too Many Requests} all say
+     * "not now", not "not ever" - langchain4j models the latter two as {@code RetriableException} upstream.
+     * They sit in the client-error family purely by HTTP numbering, which is why family alone is too blunt
+     * a discriminator for retryability.
+     */
+    private static final Set<Integer> TRANSIENT_CLIENT_ERRORS = Set.of(
+            Response.Status.REQUEST_TIMEOUT.getStatusCode(),
+            425, // Too Early - no jakarta.ws.rs.core.Response.Status constant
+            429); // Too Many Requests - no jakarta.ws.rs.core.Response.Status constant
+
+    /**
+     * Whether a provider status means "this exact request can never succeed", so the caller should give up
+     * rather than retry.
+     *
+     * <p>True only for the client-error family minus {@link #TRANSIENT_CLIENT_ERRORS}: a malformed,
+     * oversized, unauthorised or forbidden request is rejected identically however many times it is
+     * replayed. Server errors are excluded - a 5xx is the provider having a bad moment, and that is the
+     * textbook retry case. An unrecognised status is treated as retryable, matching
+     * {@code BaseRedisSubscriber}'s own "unknown defaults to retryable for safety" stance.
+     */
+    // Package-private for unit tests.
+    static boolean isPermanentFailure(int status) {
+        return familyOf(status) == Response.Status.Family.CLIENT_ERROR
+                && !TRANSIENT_CLIENT_ERRORS.contains(status);
+    }
+
     /**
      * 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

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberCursorTest.java

@@ -0,0 +1,44 @@
+package com.comet.opik.api.resources.v1.events;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.redisson.api.stream.StreamMessageId;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Pure-function coverage for {@link BaseRedisSubscriber#nextCursor(StreamMessageId)}, the mapping that
+ * decides where the next {@code XAUTOCLAIM} scan resumes (OPIK-8240).
+ *
+ * <p>Separate from {@link BaseRedisSubscriberUnitTest} because these need no Redis mocks at all, and
+ * living under that class's Mockito setup would trip strict stubbing.
+ */
+@DisplayName("BaseRedisSubscriber claim-cursor mapping")
+class BaseRedisSubscriberCursorTest {
+
+    @Test
+    @DisplayName("Redis's 0-0 end-of-pass reply wraps back to the start of the pending list")
+    void endOfPassWrapsToStart() {
+        // Compared numerically on purpose. StreamMessageId.MIN and .ALL are wire sentinels that serialize
+        // to "-" and "0", and NEITHER is equals() to the StreamMessageId(0, 0) that Redisson parses Redis's
+        // literal 0-0 into -- verified against redisson 4.7.0. Matching on the constants would therefore
+        // never fire, the cursor would stick at the end of the PEL, and the scan would stop finding
+        // anything at all: strictly worse than the always-MIN behaviour this replaces.
+        assertThat(BaseRedisSubscriber.nextCursor(new StreamMessageId(0, 0))).isEqualTo(StreamMessageId.MIN);
+    }
+
+    @Test
+    @DisplayName("A missing next-id is treated as end-of-pass rather than propagating a null")
+    void nullNextIdWrapsToStart() {
+        assertThat(BaseRedisSubscriber.nextCursor(null)).isEqualTo(StreamMessageId.MIN);
+    }
+
+    @Test
+    @DisplayName("A mid-scan position is carried forward unchanged, which is what walks past the first window")
+    void midScanPositionIsCarriedForward() {
+        // Redis caps each XAUTOCLAIM at COUNT * 10 entries EXAMINED, so without carrying this forward the
+        // scan only ever sees the first ~100 pending entries and anything behind them is never reclaimed.
+        var midScan = new StreamMessageId(1787891543627L, 0);
+        assertThat(BaseRedisSubscriber.nextCursor(midScan)).isSameAs(midScan);
+    }
+}