Merge branch 'main' into yaricom/OPIK-5555-ollie-trace-context-fix

Author: yaricom

.github/workflows/sync_provider_models.yml

@@ -85,8 +85,34 @@ jobs:
           git add apps/opik-backend/src/main/java/com/comet/opik/infrastructure/llm/
           git add apps/opik-frontend/src/types/providers.ts
           git add apps/opik-frontend/src/hooks/useLLMProviderModelsData.ts
+          git add apps/opik-backend/src/main/resources/llm-models-default.yaml
           git commit -m "[NA] [BE][FE] chore: sync provider model definitions"
 
+      - name: Configure AWS credentials
+        if: steps.sync.outputs.exit_code == '0'
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          aws-access-key-id: ${{ secrets.AWS_OPIK_CDN_KEY_ID }}
+          aws-secret-access-key: ${{ secrets.AWS_OPIK_CDN_SECRERT }}
+          aws-region: us-east-1
+          role-to-assume: ${{ vars.AWS_OPIK_CDN_ROLE }}
+          role-session-name: sync-provider-models
+
+      - name: Upload YAML to S3
+        if: steps.sync.outputs.exit_code == '0'
+        run: |
+          aws s3 cp \
+            apps/opik-backend/src/main/resources/llm-models-default.yaml \
+            ${{ vars.AWS_OPIK_CDN_BUCKET }}/llm-models-default.yaml \
+            --cache-control "max-age=300"
+
+      - name: Invalidate CloudFront distribution
+        if: steps.sync.outputs.exit_code == '0'
+        run: |
+          aws cloudfront create-invalidation \
+            --distribution-id ${{ secrets.AWS_OPIK_CDN_DISTRIBUTION }} \
+            --paths "/opik/llm-models-default.yaml"
+
       - name: Create Pull Request
         if: steps.check_changes.outputs.has_changes == 'true'
         uses: peter-evans/create-pull-request@v6

apps/opik-backend/config.yml

@@ -595,9 +595,6 @@ onlineScoring:
 
 # Configuration for Evaluation Suite assertions
 evalSuite:
-  # Default: gpt-5-nano
-  # Description: Default LLM model used for eval suite assertions when the evaluator config has no model specified
-  defaultModelName: ${EVAL_SUITE_DEFAULT_MODEL_NAME:-gpt-5-nano}
   # Default: 1
   # Description: Number of LLM runs per dataset item during eval suite execution
   defaultRunsPerItem: ${EVAL_SUITE_DEFAULT_RUNS_PER_ITEM:-1}
@@ -967,10 +964,19 @@ serviceToggles:
   # Default: false
   # Description: Whether or not Collaborators tab feature is enabled
   collaboratorsTabEnabled: ${TOGGLE_COLLABORATORS_TAB_ENABLED:-"false"}
+  # Default: "" (empty, no allowlisted workspaces)
+  # Description: Comma-separated list of workspace IDs that should always receive V2 navigation.
+  # Valid values: comma-separated workspace IDs (e.g. "ws-id-1,ws-id-2"), or "" (empty, disabled).
+  # Scope: Per-workspace — only listed workspace IDs are affected, all others follow normal determination.
+  # Priority: Highest — overrides forceWorkspaceVersion and all other determination logic for listed workspaces.
+  # Use case: gradual V2 rollout — set forceWorkspaceVersion to "version_1" and allowlist specific workspace IDs for V2.
+  # Operational impact: empty/missing means no allowlisting. Non-allowlisted workspaces are unaffected.
+  v2WorkspaceAllowlist: ${TOGGLE_V2_WORKSPACE_ALLOWLIST:-""}
   # Default: version_1
-  # Description: Forces ALL workspaces to a specific navigation version. Highest priority override.
+  # Description: Forces ALL workspaces to a specific navigation version.
   # Valid values: "disabled" (use version 1 entity check), "version_1" (force legacy), "version_2" (force project-first).
   # Scope: Global — applies to all workspaces regardless of their entity state.
+  # Priority: Second highest — overridden by v2WorkspaceAllowlist for listed workspaces.
   # Overrides: When not "disabled", overrides the version entity check and auth gate entirely.
   # Operational impact: Set to "version_1" until V2 is ready, then change to "disabled" to enable entity-based determination.
   forceWorkspaceVersion: ${TOGGLE_FORCE_WORKSPACE_VERSION:-"version_1"}

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

@@ -285,9 +285,11 @@ private void removeConsumer() {
         try {
             stream.removeConsumer(config.getConsumerGroupName(), consumerId)
                     .subscribeOn(consumerScheduler)
-                    .doOnSuccess(pendingMessages -> log.info(
-                            "Removed consumer '{}', from group '{}', pendingMessages '{}'",
-                            consumerId, config.getConsumerGroupName(), pendingMessages))
+                    .doOnSuccess(pendingMessages -> {
+                        pendingMessages = Objects.requireNonNullElse(pendingMessages, 0L);
+                        log.info("Removed consumer '{}', from group '{}', pendingMessages '{}'",
+                                consumerId, config.getConsumerGroupName(), pendingMessages);
+                    })
                     .onErrorResume(throwable -> {
                         log.warn("Failed to remove consumer '{}', group '{}'",
                                 consumerId, config.getConsumerGroupName(), throwable);
@@ -378,6 +380,7 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 .map(AutoClaimResult::getMessages)
                 .filter(Objects::nonNull)
                 .doOnSuccess(claimedMessages -> {
+                    claimedMessages = Objects.requireNonNullElse(claimedMessages, Map.of());
                     claimSize.set(claimedMessages.size());
                     log.debug("Successfully auto claimed from stream, size '{}'", claimedMessages.size());
                 })
@@ -400,7 +403,10 @@ private Mono<Map<StreamMessageId, Map<String, M>>> readMessages() {
         return stream.readGroup(config.getConsumerGroupName(), consumerId, streamReadGroupArgs)
                 .subscribeOn(consumerScheduler) // Isolates the Redis call
                 .filter(Objects::nonNull)
+                // doOnSuccess fires with null for empty Monos (e.g. long-poll timeout).
+                // Defaulting to empty map to avoid NullPointerException, and for the gauge to reset to 0
                 .doOnSuccess(messages -> {
+                    messages = Objects.requireNonNullElse(messages, Map.of());
                     readSize.set(messages.size());
                     log.debug("Successfully read from stream, size '{}'", messages.size());
                 })
@@ -434,9 +440,23 @@ private Mono<Map<StreamMessageId, Map<String, M>>> recoverFromNoGroup() {
     private Mono<ProcessingResult> processMessage(Map.Entry<StreamMessageId, Map<String, M>> entry) {
         var messageId = entry.getKey();
         log.info("Message received with messageId '{}'", messageId);
-        var message = Optional.ofNullable(entry.getValue())
-                .map(valueMap -> valueMap.get(payloadField))
-                .orElse(null);
+        M message;
+        try {
+            message = Optional.ofNullable(entry.getValue())
+                    .map(valueMap -> valueMap.get(payloadField))
+                    .orElse(null);
+        } catch (ClassCastException classCastException) {
+            // Fix for OPIK-5647: received Collections.emptyList() as the entry value for empty/malformed stream
+            // entries, which the generic Map<String, M> type erasure hides at compile time.
+            // ClassCastException is already in NON_RETRYABLE_EXCEPTIONS,
+            // so the failure path will ack and remove the message without retry.
+            // Not logging here — postProcessFailureMessages logs non-retryable errors with full context.
+            return Mono.just(ProcessingResult.builder()
+                    .messageId(messageId)
+                    .status(MessageStatus.FAILURE)
+                    .error(classCastException)
+                    .build());
+        }
         var startMillis = System.currentTimeMillis();
         // Deferring as processEvent is out of our control, it might not return a cold Mono
         return Mono.defer(() -> processEvent(message))
@@ -535,7 +555,10 @@ private Mono<Long> ackAndRemoveMessages(List<StreamMessageId> messageIds) {
                 // Only attempt to remove if ack was successful
                 .then(stream.remove(idsArray)
                         .subscribeOn(consumerScheduler))
-                .doOnSuccess(size -> log.debug("Successfully ack and remove from stream, size '{}'", size))
+                .doOnSuccess(size -> {
+                    size = Objects.requireNonNullElse(size, 0L);
+                    log.debug("Successfully ack and remove from stream, size '{}'", size);
+                })
                 .onErrorResume(throwable -> {
                     // If ack and or remove fails, message will be automatically claimed and retried
                     ackAndRemoveErrors.add(1);
@@ -614,7 +637,7 @@ private Mono<PendingEntry> listPending(StreamMessageId messageId) {
                 .subscribeOn(consumerScheduler)
                 .filter(CollectionUtils::isNotEmpty)
                 .map(List::getFirst) // Count is 1, so there would be only the first one
-                .doOnSuccess(size -> log.debug("Successfully list pending messageId '{}'", messageId))
+                .doOnNext(pendingEntry -> log.debug("Successfully list pending messageId '{}'", messageId))
                 .onErrorResume(throwable -> {
                     listPendingErrors.add(1);
                     log.warn("Error listing pending messageId '{}'", messageId, throwable);

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

@@ -2,7 +2,9 @@
 
 import com.comet.opik.api.DatasetVersion;
 import com.comet.opik.api.EvaluatorItem;
+import com.comet.opik.api.LlmProvider;
 import com.comet.opik.api.PromptType;
+import com.comet.opik.api.ProviderApiKey;
 import com.comet.opik.api.Trace;
 import com.comet.opik.api.evaluators.AutomationRuleEvaluatorType;
 import com.comet.opik.api.events.TraceToScoreLlmAsJudge;
@@ -11,6 +13,7 @@
 import com.comet.opik.domain.DatasetItemService;
 import com.comet.opik.domain.DatasetVersionService;
 import com.comet.opik.domain.IdGenerator;
+import com.comet.opik.domain.LlmProviderApiKeyService;
 import com.comet.opik.domain.evaluators.OnlineScorePublisher;
 import com.comet.opik.infrastructure.EvalSuiteConfig;
 import com.comet.opik.infrastructure.auth.RequestContext;
@@ -21,6 +24,7 @@
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.context.Context;
 import ru.vyarus.dropwizard.guice.module.installer.feature.eager.EagerSingleton;
 import ru.vyarus.dropwizard.guice.module.yaml.bind.Config;
 
@@ -30,7 +34,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
@@ -56,6 +62,7 @@ public class EvalSuiteAssertionSampler {
     private final IdGenerator idGenerator;
     private final EvalSuiteConfig evalSuiteConfig;
     private final EvalSuiteEvaluatorMapper evaluatorMapper;
+    private final LlmProviderApiKeyService llmProviderApiKeyService;
 
     @Inject
     public EvalSuiteAssertionSampler(
@@ -64,13 +71,15 @@ public EvalSuiteAssertionSampler(
             @NonNull OnlineScorePublisher onlineScorePublisher,
             @NonNull IdGenerator idGenerator,
             @NonNull @Config("evalSuite") EvalSuiteConfig evalSuiteConfig,
-            @NonNull EvalSuiteEvaluatorMapper evaluatorMapper) {
+            @NonNull EvalSuiteEvaluatorMapper evaluatorMapper,
+            @NonNull LlmProviderApiKeyService llmProviderApiKeyService) {
         this.datasetItemService = datasetItemService;
         this.datasetVersionService = datasetVersionService;
         this.onlineScorePublisher = onlineScorePublisher;
         this.idGenerator = idGenerator;
         this.evalSuiteConfig = evalSuiteConfig;
         this.evaluatorMapper = evaluatorMapper;
+        this.llmProviderApiKeyService = llmProviderApiKeyService;
     }
 
     @Subscribe
@@ -83,13 +92,26 @@ public void onTracesCreated(TracesCreated tracesBatch) {
             return;
         }
 
-        var reactiveContext = reactor.util.context.Context.of(
+        var reactiveContext = Context.of(
                 RequestContext.WORKSPACE_ID, tracesBatch.workspaceId(),
                 RequestContext.USER_NAME, tracesBatch.userName(),
                 RequestContext.VISIBILITY, com.comet.opik.api.Visibility.PRIVATE);
 
         Duration fetchTimeout = Duration.ofSeconds(evalSuiteConfig.getFetchTimeoutSeconds());
 
+        // Resolve model once per batch: prefer connected provider, fall back to first trace's model
+        var connectedProviders = getConnectedProviders(tracesBatch.workspaceId());
+        String modelName = SupportedJudgeProvider.resolveModel(connectedProviders)
+                .or(() -> getMetadataString(completeTraces.getFirst(), "eval_suite_model"))
+                .orElse(null);
+
+        if (modelName == null) {
+            log.warn("No LLM model resolved for eval suite batch in workspace '{}' — "
+                    + "no supported provider connected and no eval_suite_model in trace metadata",
+                    tracesBatch.workspaceId());
+            return;
+        }
+
         // Cache dataset evaluators by (datasetId:versionHash) to avoid redundant fetches
         Map<String, List<PreparedEvaluator>> datasetEvaluatorsCache = new HashMap<>();
 
@@ -120,7 +142,7 @@ public void onTracesCreated(TracesCreated tracesBatch) {
                                 .contextWrite(reactiveContext)
                                 .timeout(fetchTimeout)
                                 .block();
-                        return evaluatorMapper.prepareEvaluators(result.evaluators());
+                        return evaluatorMapper.prepareEvaluators(result.evaluators(), modelName);
                     });
 
                     var datasetItemId = getMetadataString(trace, "eval_suite_dataset_item_id");
@@ -134,7 +156,8 @@ public void onTracesCreated(TracesCreated tracesBatch) {
                             .flatMap(itemId -> {
                                 List<PreparedEvaluator> allEvaluators = new ArrayList<>(
                                         preparedDatasetEvaluators);
-                                allEvaluators.addAll(fetchItemEvaluators(itemId, reactiveContext));
+                                allEvaluators.addAll(fetchItemEvaluators(itemId, reactiveContext,
+                                        modelName));
 
                                 if (allEvaluators.isEmpty()) {
                                     log.debug("No evaluators found for trace '{}', dataset item '{}'",
@@ -191,7 +214,8 @@ private Mono<DatasetEvaluatorsResult> fetchDatasetEvaluators(UUID datasetId, Str
     }
 
     private List<PreparedEvaluator> fetchItemEvaluators(
-            UUID itemId, reactor.util.context.Context reactiveContext) {
+            UUID itemId, Context reactiveContext,
+            String modelName) {
         try {
             var item = datasetItemService.get(itemId)
                     .contextWrite(reactiveContext)
@@ -202,7 +226,7 @@ private List<PreparedEvaluator> fetchItemEvaluators(
                 return List.of();
             }
 
-            return evaluatorMapper.prepareEvaluators(item.evaluators());
+            return evaluatorMapper.prepareEvaluators(item.evaluators(), modelName);
         } catch (Exception e) {
             log.error("Failed to fetch evaluators for item '{}'", itemId, e);
             return List.of();
@@ -230,4 +254,16 @@ private Optional<UUID> parseUUID(String id, UUID traceId) {
         }
     }
 
+    private Set<LlmProvider> getConnectedProviders(String workspaceId) {
+        try {
+            return llmProviderApiKeyService.find(workspaceId)
+                    .content().stream()
+                    .map(ProviderApiKey::provider)
+                    .collect(Collectors.toSet());
+        } catch (Exception e) {
+            log.error("Failed to fetch connected providers for workspace '{}'", workspaceId, e);
+            return Set.of();
+        }
+    }
+
 }

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

@@ -15,13 +15,11 @@
 import jakarta.inject.Singleton;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -54,7 +52,8 @@ public int getEffectiveRunsPerItem(ExecutionPolicy itemPolicy, ExecutionPolicy v
         return evalSuiteConfig.getDefaultRunsPerItem();
     }
 
-    public List<PreparedEvaluator> prepareEvaluators(List<EvaluatorItem> evaluators) {
+    public List<PreparedEvaluator> prepareEvaluators(List<EvaluatorItem> evaluators,
+            String modelName) {
         return evaluators.stream()
                 .filter(evaluator -> {
                     if (evaluator.type() != EvaluatorType.LLM_JUDGE) {
@@ -66,7 +65,7 @@ public List<PreparedEvaluator> prepareEvaluators(List<EvaluatorItem> evaluators)
                 })
                 .flatMap(evaluator -> {
                     try {
-                        LlmAsJudgeCode code = toScoringCode(evaluator.config());
+                        LlmAsJudgeCode code = toScoringCode(evaluator.config(), modelName);
 
                         Map<String, String> scoreNameMapping = code.schema() != null
                                 ? code.schema().stream()
@@ -84,16 +83,22 @@ public List<PreparedEvaluator> prepareEvaluators(List<EvaluatorItem> evaluators)
                 .toList();
     }
 
-    LlmAsJudgeCode toScoringCode(JsonNode config) {
-        LlmAsJudgeCode code = deserializeEvaluatorConfig(config);
-        code = resolveModelName(code);
+    LlmAsJudgeCode toScoringCode(JsonNode config, String modelName) {
+        LlmAsJudgeCode code = deserializeScoringCode(config, modelName);
         code = renameSchemaToAssertionKeys(code);
         code = applyEvalSuitePrompt(code);
         return code;
     }
 
-    private LlmAsJudgeCode deserializeEvaluatorConfig(JsonNode config) {
-        return JsonUtils.treeToValue(config, LlmAsJudgeCode.class);
+    private LlmAsJudgeCode deserializeScoringCode(JsonNode config, String modelName) {
+        var code = JsonUtils.treeToValue(config, LlmAsJudgeCode.class);
+        var existingModel = code.model();
+        var model = (existingModel != null ? existingModel.toBuilder() : LlmAsJudgeModelParameters.builder())
+                .name(modelName)
+                .build();
+        return code.toBuilder()
+                .model(model)
+                .build();
     }
 
     /**
@@ -160,17 +165,4 @@ private String formatAssertions(List<LlmAsJudgeOutputSchema> schema) {
                 .collect(Collectors.joining("\n"));
     }
 
-    private LlmAsJudgeCode resolveModelName(LlmAsJudgeCode code) {
-        var existingModel = Optional.ofNullable(code.model());
-        if (existingModel.map(LlmAsJudgeModelParameters::name).filter(StringUtils::isNotBlank).isEmpty()) {
-            var resolvedModel = LlmAsJudgeModelParameters.builder()
-                    .name(evalSuiteConfig.getDefaultModelName())
-                    .temperature(existingModel.map(LlmAsJudgeModelParameters::temperature).orElse(null))
-                    .seed(existingModel.map(LlmAsJudgeModelParameters::seed).orElse(null))
-                    .customParameters(existingModel.map(LlmAsJudgeModelParameters::customParameters).orElse(null))
-                    .build();
-            return new LlmAsJudgeCode(resolvedModel, code.messages(), code.variables(), code.schema());
-        }
-        return code;
-    }
 }