[OPIK-8183] [BE] feat: accept CIPX device tokens and record device_id on cipx identities

apps/opik-backend/config.yml

@@ -394,6 +394,17 @@ mcpOAuth:
   # Description: Maximum time the scrub job waits to acquire the Redis lock before giving up for this cycle
   scrubLockWaitTime: ${MCP_OAUTH_SCRUB_LOCK_WAIT_TIME:-PT0.1S}
 
+# Delegated validation of CIPX device tokens (Authorization: opik_cipx_at_...). Validation, revocation and the
+# device registry live in cost-api, so no key material or crypto is needed here. Disabled by default: while
+# disabled the authentication filter never inspects the prefix and API-key auth is untouched.
+cipxTokenValidation:
+  # Default: false
+  # Description: Whether to accept CIPX device tokens as a credential
+  enabled: ${CIPX_TOKEN_VALIDATION_ENABLED:-false}
+  # Default: http://ai-cost-backend
+  # Description: Base URL of the cost-api service that validates device tokens; the validator endpoint is appended to it. The in-cluster service name, not the public front door: the validator is reached only from inside the cluster
+  url: ${CIPX_TOKEN_VALIDATION_URL:-http://ai-cost-backend}
+
 # https://www.dropwizard.io/en/stable/manual/configuration.html#servers
 server:
   # Default: false

apps/opik-backend/src/main/java/com/comet/opik/api/events/TraceCostIntelligenceChanged.java

@@ -2,6 +2,7 @@
 
 import com.comet.opik.api.TraceUpdate;
 import com.comet.opik.infrastructure.events.BaseEvent;
+import jakarta.annotation.Nullable;
 import lombok.Getter;
 import lombok.NonNull;
 import lombok.experimental.Accessors;
@@ -21,11 +22,18 @@ public class TraceCostIntelligenceChanged extends BaseEvent {
 
     private final @NonNull Map<UUID, UUID> traceProjectIds;
     private final @NonNull TraceUpdate traceUpdate;
+    private final @Nullable String cipxDeviceId;
 
     public TraceCostIntelligenceChanged(@NonNull Map<UUID, UUID> traceProjectIds, @NonNull TraceUpdate traceUpdate,
             @NonNull String workspaceId, @NonNull String userName) {
+        this(traceProjectIds, traceUpdate, workspaceId, userName, null);
+    }
+
+    public TraceCostIntelligenceChanged(@NonNull Map<UUID, UUID> traceProjectIds, @NonNull TraceUpdate traceUpdate,
+            @NonNull String workspaceId, @NonNull String userName, @Nullable String cipxDeviceId) {
         super(workspaceId, userName);
         this.traceProjectIds = traceProjectIds;
         this.traceUpdate = traceUpdate;
+        this.cipxDeviceId = cipxDeviceId;
     }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/events/TracesCreated.java

@@ -19,16 +19,23 @@ public class TracesCreated extends BaseEvent {
     // Resolved from RequestContext.WORKSPACE_NAME at publish time (TraceService). May be null/blank
     // for callers that don't carry it; consumers fall back to workspaceId.
     private final @Nullable String workspaceName;
+    private final @Nullable String cipxDeviceId;
 
     public TracesCreated(@NonNull List<Trace> traces, @NonNull String workspaceId, @NonNull String userName) {
         this(traces, workspaceId, userName, null);
     }
 
     public TracesCreated(@NonNull List<Trace> traces, @NonNull String workspaceId, @NonNull String userName,
             @Nullable String workspaceName) {
+        this(traces, workspaceId, userName, workspaceName, null);
+    }
+
+    public TracesCreated(@NonNull List<Trace> traces, @NonNull String workspaceId, @NonNull String userName,
+            @Nullable String workspaceName, @Nullable String cipxDeviceId) {
         super(workspaceId, userName);
         this.traces = traces;
         this.workspaceName = workspaceName;
+        this.cipxDeviceId = cipxDeviceId;
     }
 
     public Set<UUID> projectIds() {

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

@@ -35,8 +35,10 @@
  * update, so identity create reuses TracesCreated (which carries the full entities) and update
  * consumes the dedicated TraceCostIntelligenceChanged event (TracesUpdated carries only a delta).
  * cipx spans/traces are filtered out here in Java before any DB work, and all cipx fields are
- * derived in Java. Runs on the AsyncEventBus virtual threads, off the request path; failures are
- * logged and swallowed — ingestion of the source span/trace already succeeded.
+ * derived in Java, except device_id: that comes from the event, which carries the value resolved from
+ * the caller's validated device token, so a client cannot claim another machine's identity by writing
+ * one into its own metadata. Runs on the AsyncEventBus virtual threads, off the request path; failures
+ * are logged and swallowed — ingestion of the source span/trace already succeeded.
  */
 @EagerSingleton
 @Slf4j
@@ -109,7 +111,7 @@ public void onTracesCreated(TracesCreated event) {
         List<TraceIdentityRow> rows = event.traces().stream()
                 .filter(trace -> CipxMetadata.hasIdentity(trace.metadata()))
                 .map(trace -> TraceIdentityRow.from(trace.id(), trace.projectId(), trace.metadata(),
-                        trace.startTime()))
+                        trace.startTime(), event.cipxDeviceId()))
                 .toList();
         ingestIdentities(rows, event.workspaceId(), event.userName());
     }
@@ -132,7 +134,8 @@ public void onTraceCostIntelligenceChanged(TraceCostIntelligenceChanged event) {
                             List<TraceIdentityRow> rows = traceProjectIds.entrySet().stream()
                                     .filter(entry -> startTimes.containsKey(entry.getKey()))
                                     .map(entry -> TraceIdentityRow.from(entry.getKey(), entry.getValue(),
-                                            update.metadata(), startTimes.get(entry.getKey())))
+                                            update.metadata(), startTimes.get(entry.getKey()),
+                                            event.cipxDeviceId()))
                                     .toList();
                             ingestIdentities(rows, event.workspaceId(), event.userName());
                         },

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

@@ -13,6 +13,7 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.reactivestreams.Publisher;
 import org.stringtemplate.v4.ST;
 import reactor.core.publisher.Mono;
@@ -49,6 +50,7 @@ public record TraceIdentityRow(
             @NonNull String repository,
             @NonNull String sessionId,
             @NonNull String harness,
+            @NonNull String deviceId,
             int schemaVersion,
             @NonNull String billingMode,
             @NonNull String plan,
@@ -66,7 +68,8 @@ public record TraceIdentityRow(
             int linesAdded,
             int linesDeleted) {
 
-        public static TraceIdentityRow from(UUID traceId, UUID projectId, JsonNode metadata, Instant startTime) {
+        public static TraceIdentityRow from(UUID traceId, UUID projectId, JsonNode metadata, Instant startTime,
+                String deviceId) {
             JsonNode session = metadata.path("cipx").path("session");
             JsonNode identity = session.path("identity");
             JsonNode repository = session.path("repository");
@@ -84,6 +87,7 @@ public static TraceIdentityRow from(UUID traceId, UUID projectId, JsonNode metad
                     .repository(repository.path("remote").asText(""))
                     .sessionId(session.path("session_id").asText(""))
                     .harness(session.path("harness").asText(""))
+                    .deviceId(StringUtils.defaultString(deviceId))
                     .schemaVersion(session.path("schema_version").asInt(0))
                     .billingMode(identity.path("billing_mode").asText(""))
                     .plan(identity.path("plan").asText(""))
@@ -109,7 +113,7 @@ public static TraceIdentityRow from(UUID traceId, UUID projectId, JsonNode metad
     private static final String INSERT = """
             INSERT INTO cipx_trace_identities
                 (workspace_id, project_id, trace_id, start_time, user_uuid,
-                 user_email, user_display_name, repository, session_id, harness, schema_version,
+                 user_email, user_display_name, repository, session_id, harness, device_id, schema_version,
                  billing_mode, plan, plan_usage_status, organization_type, seat_tier, billing_type,
                  branch, head_sha_start, head_sha_end, dirty, commits_in_trace,
                  files_added, files_deleted, lines_added, lines_deleted)
@@ -127,6 +131,7 @@ public static TraceIdentityRow from(UUID traceId, UUID projectId, JsonNode metad
                         :repository<item.index>,
                         :session_id<item.index>,
                         :harness<item.index>,
+                        :device_id<item.index>,
                         :schema_version<item.index>,
                         :billing_mode<item.index>,
                         :plan<item.index>,
@@ -172,7 +177,7 @@ private Publisher<? extends Result> insert(List<TraceIdentityRow> rows, String w
         // Positional binds: the driver resolves named binds with a linear indexOf over the statement's
         // parameter list (quadratic per statement), while bind(int) is a direct array write. Indices
         // follow the placeholders' first-appearance order in the rendered SQL: workspace_id once at 0
-        // (repeats dedup), then 25 parameters per row tuple in template order.
+        // (repeats dedup), then 26 parameters per row tuple in template order.
         statement.bind(0, workspaceId);
         int index = 1;
         for (TraceIdentityRow row : rows) {
@@ -185,6 +190,7 @@ private Publisher<? extends Result> insert(List<TraceIdentityRow> rows, String w
                     .bind(index++, row.repository())
                     .bind(index++, row.sessionId())
                     .bind(index++, row.harness())
+                    .bind(index++, row.deviceId())
                     .bind(index++, row.schemaVersion())
                     .bind(index++, row.billingMode())
                     .bind(index++, row.plan())

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

@@ -150,6 +150,7 @@ public Mono<UUID> create(@NonNull Trace trace) {
                 .flatMap(project -> {
                     String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
                     String workspaceName = ctx.getOrDefault(RequestContext.WORKSPACE_NAME, "");
+                    String cipxDeviceId = ctx.getOrDefault(RequestContext.CIPX_DEVICE_ID, "");
                     String userName = ctx.get(RequestContext.USER_NAME);
 
                     // Strip attachments from the trace with the generated ID and project ID
@@ -163,7 +164,7 @@ public Mono<UUID> create(@NonNull Trace trace) {
                                         var savedTrace = processedTrace.toBuilder().projectId(project.id())
                                                 .projectName(projectName).build();
                                         eventBus.post(new TracesCreated(List.of(savedTrace), workspaceId, userName,
-                                                workspaceName));
+                                                workspaceName, cipxDeviceId));
                                     }));
                 }));
     }
@@ -205,6 +206,7 @@ public Mono<Long> create(TraceBatch batch) {
                 .then(Mono.deferContextual(ctx -> {
                     String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
                     String workspaceName = ctx.getOrDefault(RequestContext.WORKSPACE_NAME, "");
+                    String cipxDeviceId = ctx.getOrDefault(RequestContext.CIPX_DEVICE_ID, "");
                     String userName = ctx.get(RequestContext.USER_NAME);
 
                     Mono<List<Trace>> resolveProjects = Flux.fromIterable(projectNames)
@@ -222,7 +224,7 @@ public Mono<Long> create(TraceBatch batch) {
                                     .nonTransaction(connection -> dao.batchInsert(traces, connection))
                                     .doOnSuccess(__ -> {
                                         eventBus.post(new TracesCreated(traces, workspaceId, userName,
-                                                workspaceName));
+                                                workspaceName, cipxDeviceId));
                                     }));
                 }));
     }
@@ -354,7 +356,8 @@ public Mono<Void> update(@NonNull TraceUpdate traceUpdate, @NonNull UUID id) {
                                         .doOnSuccess(__ -> eventBus.post(new TraceCostIntelligenceChanged(
                                                 Map.of(id, project.id()), traceUpdate,
                                                 ctx.get(RequestContext.WORKSPACE_ID),
-                                                ctx.get(RequestContext.USER_NAME)))))))
+                                                ctx.get(RequestContext.USER_NAME),
+                                                ctx.getOrDefault(RequestContext.CIPX_DEVICE_ID, "")))))))
                         .then()));
     }
 
@@ -368,6 +371,7 @@ public Mono<Void> batchUpdate(@NonNull TraceBatchUpdate batchUpdate) {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
             String userName = ctx.get(RequestContext.USER_NAME);
             String workspaceName = ctx.getOrDefault(RequestContext.WORKSPACE_NAME, "");
+            String cipxDeviceId = ctx.getOrDefault(RequestContext.CIPX_DEVICE_ID, "");
             return dao.getProjectIdsByTraceIds(new ArrayList<>(batchUpdate.ids()))
                     .flatMap(traceToProjectMap -> {
                         var projectIds = Set.copyOf(traceToProjectMap.values());
@@ -378,7 +382,7 @@ public Mono<Void> batchUpdate(@NonNull TraceBatchUpdate batchUpdate) {
                                     eventBus.post(new TracesUpdated(projectIds, batchUpdate.ids(), workspaceId,
                                             userName, batchUpdate.update(), workspaceName, traceToProjectMap));
                                     eventBus.post(new TraceCostIntelligenceChanged(traceToProjectMap,
-                                            batchUpdate.update(), workspaceId, userName));
+                                            batchUpdate.update(), workspaceId, userName, cipxDeviceId));
                                 });
                     });
         });

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/CipxTokenValidationConfig.java

@@ -0,0 +1,24 @@
+package com.comet.opik.infrastructure;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.Valid;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Delegated validation of CIPX device tokens. Disabled by default.
+ */
+@Data
+@Builder(toBuilder = true)
+@NoArgsConstructor
+@AllArgsConstructor
+public class CipxTokenValidationConfig {
+
+    @Valid @JsonProperty
+    private boolean enabled;
+
+    @Valid @JsonProperty
+    private String url;
+}

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/OpikConfiguration.java

@@ -42,6 +42,9 @@ public class OpikConfiguration extends JobConfiguration {
     @Valid @NotNull @JsonProperty
     private McpOAuthConfig mcpOAuth = new McpOAuthConfig();
 
+    @Valid @NotNull @JsonProperty
+    private CipxTokenValidationConfig cipxTokenValidation = new CipxTokenValidationConfig();
+
     @Valid @NotNull @JsonProperty
     private RedisConfig redis = new RedisConfig();
 

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/auth/AuthCredentialsCacheService.java

@@ -5,6 +5,7 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.redisson.api.RBatchReactive;
 import org.redisson.api.RBucketsReactive;
 import org.redisson.api.RMapReactive;
@@ -35,13 +36,15 @@ class AuthCredentialsCacheService implements CacheService {
     private static final String WORKSPACE_NAME_KEY = "workspaceName";
     private static final String QUOTAS_KEY = "quotas";
     private static final String PERMISSIONS_KEY = "permissions";
+    private static final String DEVICE_ID_KEY = "deviceId";
 
     private static final Set<String> V2_MAP_FIELDS = Set.of(
             USER_NAME_KEY,
             WORKSPACE_ID_KEY,
             WORKSPACE_NAME_KEY,
             QUOTAS_KEY,
-            PERMISSIONS_KEY);
+            PERMISSIONS_KEY,
+            DEVICE_ID_KEY);
 
     private final @NonNull RedissonReactiveClient redissonClient;
     private final int ttlInSeconds;
@@ -68,6 +71,9 @@ private Optional<AuthCredentials> resolveFromV2Cache(String apiKey, String works
                         .workspaceName(m.get(WORKSPACE_NAME_KEY))
                         .quotas(getQuotas(m))
                         .permissions(getPermissions(m))
+                        // Blank for every credential that is not a CIPX device token, and absent for entries
+                        // written before device ids; both mean no device, so they read back the same.
+                        .deviceId(StringUtils.trimToNull(m.get(DEVICE_ID_KEY)))
                         .build());
     }
 
@@ -103,7 +109,8 @@ public void cache(
                 WORKSPACE_NAME_KEY, Optional.ofNullable(credentials.workspaceName()).orElse(requestWorkspaceName),
                 QUOTAS_KEY, JsonUtils.writeValueAsString(Optional.ofNullable(credentials.quotas()).orElse(List.of())),
                 PERMISSIONS_KEY,
-                JsonUtils.writeValueAsString(Optional.ofNullable(credentials.permissions()).orElse(List.of())));
+                JsonUtils.writeValueAsString(Optional.ofNullable(credentials.permissions()).orElse(List.of())),
+                DEVICE_ID_KEY, StringUtils.defaultString(credentials.deviceId()));
 
         RBatchReactive batch = redissonClient.createBatch();
         RMapReactive<String, String> v2Map = batch.getMap(v2Key);

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/auth/AuthFilter.java

@@ -33,6 +33,7 @@ public class AuthFilter implements ContainerRequestFilter {
 
     private final AuthService authService;
     private final McpOAuthService mcpOAuthService;
+    private final CipxTokenValidationService cipxTokenValidationService;
     private final OpikConfiguration opikConfig;
     private final Provider<RequestContext> requestContext;
 
@@ -62,6 +63,8 @@ public void filter(ContainerRequestContext context) throws IOException {
                 ValidatedToken validatedToken = mcpOAuthService.validateAccessTokenForWorkspace(
                         token, context.getHeaderString(RequestContext.WORKSPACE_HEADER));
                 authService.authorizeOAuth(validatedToken, contextInfo);
+            } else if (opikConfig.getCipxTokenValidation().isEnabled() && CipxTokenUtils.isCipxToken(authHeader)) {
+                cipxTokenValidationService.authenticate(authHeader, contextInfo);
             } else {
                 authService.authenticate(headers, sessionToken, contextInfo);
             }