[OPIK-7352] [BE] reject non-UUIDv7 referenced ids on ingest

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

@@ -555,6 +555,8 @@ private Alert prepareAlert(Alert alert, String userName, String workspaceId) {
 
         UUID id = alert.id() == null ? idGenerator.generateId() : alert.id();
         IdGenerator.validateVersion(id, "Alert");
+        // projectId is persisted without an existence check here, so enforce v7 to avoid storing an orphan v4.
+        idGenerator.validateIdNotInFutureIfPresent(alert.projectId(), "project");
 
         UUID webhookId = alert.webhook().id() == null ? idGenerator.generateId() : alert.webhook().id();
         IdGenerator.validateVersion(webhookId, "Webhook");

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

@@ -134,6 +134,10 @@ public Mono<Long> addItems(@NonNull UUID queueId, @NonNull Set<UUID> itemIds) {
             return Mono.just(0L);
         }
 
+        // Queue items reference trace/thread ids (v7 by construction); enforce so the referenced-id
+        // policy is uniform. Past allowed — queues commonly collect older traces/threads.
+        itemIds.forEach(itemId -> idGenerator.validateIdNotInFuture(itemId, "AnnotationQueue item"));
+
         return annotationQueueDAO.findQueueInfoById(queueId)
                 .switchIfEmpty(Mono.error(createNotFoundError(queueId)))
                 .flatMap(queue -> annotationQueueDAO.addItems(queueId, itemIds, queue.projectId()))
@@ -257,6 +261,8 @@ private Mono<AnnotationQueue.AnnotationQueuePage> enhancePageWithProjectNames(
     private AnnotationQueue prepareAnnotationQueue(AnnotationQueue annotationQueue) {
         UUID id = annotationQueue.id() == null ? idGenerator.generateId() : annotationQueue.id();
         IdGenerator.validateVersion(id, "AnnotationQueue");
+        // projectId is persisted without an existence check here, so enforce v7 to avoid storing an orphan v4.
+        idGenerator.validateIdNotInFutureIfPresent(annotationQueue.projectId(), "project");
 
         log.debug("Preparing annotation queue with id '{}', name '{}', project '{}'",
                 id, annotationQueue.name(), annotationQueue.projectId());

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

@@ -43,6 +43,7 @@ class AssertionResultServiceImpl implements AssertionResultService {
     private final @NonNull AssertionResultDAO assertionResultDAO;
     private final @NonNull ProjectService projectService;
     private final @NonNull EventBus eventBus;
+    private final @NonNull IdGenerator idGenerator;
 
     @Override
     public Mono<Long> insertBatch(@NonNull EntityType entityType,
@@ -63,7 +64,7 @@ public Mono<Void> saveBatch(@NonNull EntityType entityType,
         }
 
         // Validate up front so a bad id fails fast and independently of project-name normalisation.
-        assertionResults.forEach(item -> IdGenerator.validateVersion(item.entityId(), entityType.getType()));
+        assertionResults.forEach(item -> idGenerator.validateIdNotInFuture(item.entityId(), entityType.getType()));
 
         return Mono.deferContextual(ctx -> {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);

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

@@ -53,6 +53,7 @@ class CommentServiceImpl implements CommentService {
 
     @Override
     public Mono<UUID> create(@NonNull UUID entityId, @NonNull Comment comment, CommentDAO.EntityType entityType) {
+        idGenerator.validateIdNotInFuture(entityId, entityType.getType());
         UUID id = idGenerator.generateId();
         var monoProjectId = resolveProjectId(entityType, entityId);
 

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

@@ -187,6 +187,8 @@ public Mono<Void> createFromTraces(
 
         log.info("Creating dataset items from '{}' traces for dataset '{}'", traceIds.size(), datasetId);
 
+        traceIds.forEach(traceId -> idGenerator.validateIdNotInFuture(traceId, "dataset_item trace"));
+
         // Verify dataset exists
         return Mono.deferContextual(ctx -> {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
@@ -240,6 +242,8 @@ public Mono<Void> createFromSpans(
 
         log.info("Creating dataset items from '{}' spans for dataset '{}'", spanIds.size(), datasetId);
 
+        spanIds.forEach(spanId -> idGenerator.validateIdNotInFuture(spanId, "dataset_item span"));
+
         // Verify dataset exists
         return Mono.deferContextual(ctx -> {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
@@ -382,6 +386,7 @@ private Mono<DatasetItem> authorizeItem(Mono<DatasetItem> itemMono) {
     @Override
     @WithSpan
     public Mono<Void> patch(@NonNull UUID id, @NonNull DatasetItem item) {
+        validateReferencedTraceAndSpan(item);
         return Mono.deferContextual(ctx -> {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
             String userName = ctx.get(RequestContext.USER_NAME);
@@ -903,11 +908,23 @@ private List<DatasetItem> addIdIfAbsent(DatasetItemBatch batch) {
                 .stream()
                 .map(item -> {
                     IdGenerator.validateVersion(item.id(), "dataset_item");
+                    validateReferencedTraceAndSpan(item);
                     return item;
                 })
                 .toList();
     }
 
+    // The dataset_item's referenced trace_id / span_id must be a time-ordered UUIDv7 (past allowed:
+    // items are commonly linked to older traces/spans). Reuses the shared referenced-id policy.
+    private void validateReferencedTraceAndSpan(DatasetItem item) {
+        if (item.traceId() != null) {
+            idGenerator.validateIdNotInFuture(item.traceId(), "dataset_item trace");
+        }
+        if (item.spanId() != null) {
+            idGenerator.validateIdNotInFuture(item.spanId(), "dataset_item span");
+        }
+    }
+
     private <T> Mono<T> failWithConflict(String message) {
         log.info(message);
         return Mono.error(new IdentifierMismatchException(new ErrorMessage(List.of(message))));

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

@@ -495,6 +495,8 @@ private Set<UUID> getPromptVersionIds(Experiment experiment) {
     public Mono<UUID> create(@NonNull Experiment experiment) {
         var id = experiment.id() == null ? idGenerator.generateId() : experiment.id();
         IdGenerator.validateVersion(id, "Experiment");
+        // optimizationId is stored without an existence check, so enforce v7 to avoid storing an orphan v4.
+        idGenerator.validateIdNotInFutureIfPresent(experiment.optimizationId(), "optimization");
         var name = StringUtils.getIfBlank(experiment.name(), nameGenerator::generateName);
         return resolveProjectId(experiment)
                 .flatMap(resolvedExperiment -> datasetService

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

@@ -89,6 +89,7 @@ class FeedbackScoreServiceImpl implements FeedbackScoreService {
     private final @NonNull TraceThreadService traceThreadService;
     private final @NonNull Provider<RequestContext> requestContext;
     private final @NonNull EventBus eventBus;
+    private final @NonNull IdGenerator idGenerator;
 
     @Builder(toBuilder = true)
     record ProjectDto<T extends FeedbackScoreItem>(Project project, List<T> scores) {
@@ -100,6 +101,8 @@ public Mono<Void> scoreTrace(@NonNull UUID traceId, @NonNull FeedbackScore score
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
             String userName = ctx.get(RequestContext.USER_NAME);
 
+            idGenerator.validateIdNotInFuture(traceId, EntityType.TRACE.getType());
+            idGenerator.validateIdNotInFutureIfPresent(score.sourceQueueId(), "annotation queue");
             return traceDAO.getProjectIdFromTrace(traceId)
                     .switchIfEmpty(Mono.error(failWithNotFound("Trace", traceId)))
                     .flatMap(projectId -> getAuthor()
@@ -118,6 +121,8 @@ public Mono<Void> scoreSpan(@NonNull UUID spanId, @NonNull FeedbackScore score)
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
             String userName = ctx.get(RequestContext.USER_NAME);
 
+            idGenerator.validateIdNotInFuture(spanId, EntityType.SPAN.getType());
+            idGenerator.validateIdNotInFutureIfPresent(score.sourceQueueId(), "annotation queue");
             return spanDAO.getProjectIdFromSpan(spanId)
                     .switchIfEmpty(Mono.error(failWithNotFound("Span", spanId)))
                     .flatMap(projectId -> getAuthor()
@@ -173,7 +178,8 @@ private Mono<Void> processScoreBatch(EntityType entityType, List<FeedbackScoreBa
         Map<String, List<FeedbackScoreItem>> scoresPerProject = scores
                 .stream()
                 .map(score -> {
-                    IdGenerator.validateVersion(score.id(), entityType.getType()); // validate span/trace id
+                    idGenerator.validateIdNotInFuture(score.id(), entityType.getType()); // validate span/trace id
+                    idGenerator.validateIdNotInFutureIfPresent(score.sourceQueueId(), "annotation queue");
 
                     return score.toBuilder()
                             .projectName(WorkspaceUtils.getProjectName(score.projectName()))

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

@@ -51,7 +51,8 @@ public Mono<Void> addTraceGuardrails(List<Guardrail> guardrails) {
                 .stream()
                 .map(guardrail -> {
                     UUID id = idGenerator.generateId();
-                    IdGenerator.validateVersion(guardrail.entityId(), entityType.getType()); // validate trace id
+                    idGenerator.validateIdNotInFuture(guardrail.entityId(), entityType.getType());
+                    idGenerator.validateIdNotInFuture(guardrail.secondaryId(), "guardrail secondary");
 
                     return guardrail.toBuilder()
                             .id(id)

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

@@ -34,12 +34,27 @@ public interface IdGenerator {
     Mono<UUID> validateIdAsync(UUID id, String resource);
 
     /**
-     * Validates an ingested {@code id} on the update path: it must be a version 7 UUID
-     * ({@link #validateVersion(UUID, String)}) and must not embed a timestamp far in the future (which
-     * would corrupt the partition layout). Unlike {@link #validateId}, old ids are allowed, because
-     * updating a long-lived entity (e.g. created months ago) is a legitimate operation.
+     * Validates an {@code id} that may legitimately point at an entity created in the past: it must be a
+     * version 7 UUID ({@link #validateVersion(UUID, String)}) and must not embed a timestamp far in the
+     * future (which would corrupt the partition layout / retention id-range). Unlike {@link #validateId},
+     * old ids are allowed.
+     *
+     * <p>Used both on the update path (updating a long-lived entity created months ago is legitimate) and
+     * for referenced/foreign ids on ingest (e.g. a span's {@code traceId}: retention orders spans by the
+     * {@code trace_id} range assuming it is a time-ordered UUIDv7, and late spans on old traces are common,
+     * so old is fine but non-v7 or future-dated must be rejected).
      */
-    Mono<UUID> validateIdForUpdateAsync(UUID id, String resource);
+    void validateIdNotInFuture(UUID id, String resource);
+
+    Mono<UUID> validateIdNotInFutureAsync(UUID id, String resource);
+
+    /**
+     * Null-safe variant of {@link #validateIdNotInFuture} for optional referenced ids (e.g. an optional
+     * {@code projectId} that may be resolved by name instead). No-op when {@code id} is null.
+     */
+    void validateIdNotInFutureIfPresent(UUID id, String resource);
+
+    Mono<UUID> validateIdNotInFutureIfPresentAsync(UUID id, String resource);
 
     static Mono<UUID> validateVersionAsync(@NonNull UUID id, String resource) {
         return Mono.fromCallable(() -> {
@@ -91,16 +106,29 @@ public Mono<UUID> validateIdAsync(@NonNull UUID id, String resource) {
         });
     }
 
-    private void validateIdForUpdate(UUID id, String resource) {
+    @Override
+    public void validateIdNotInFuture(@NonNull UUID id, String resource) {
         IdGenerator.validateVersion(id, resource);
         uuidV7TimestampValidator.validateNotInFuture(id);
     }
 
     @Override
-    public Mono<UUID> validateIdForUpdateAsync(@NonNull UUID id, String resource) {
+    public Mono<UUID> validateIdNotInFutureAsync(@NonNull UUID id, String resource) {
         return Mono.fromCallable(() -> {
-            validateIdForUpdate(id, resource);
+            validateIdNotInFuture(id, resource);
             return id;
         });
     }
+
+    @Override
+    public void validateIdNotInFutureIfPresent(UUID id, String resource) {
+        if (id != null) {
+            validateIdNotInFuture(id, resource);
+        }
+    }
+
+    @Override
+    public Mono<UUID> validateIdNotInFutureIfPresentAsync(UUID id, String resource) {
+        return id == null ? Mono.empty() : validateIdNotInFutureAsync(id, resource);
+    }
 }

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

@@ -62,6 +62,8 @@ public class SpanService {
     public static final String PARENT_SPAN_IS_MISMATCH = "parent_span_id does not match the existing span";
     public static final String TRACE_ID_MISMATCH = "trace_id does not match the existing span";
     public static final String SPAN_KEY = "Span";
+    public static final String SPAN_TRACE_KEY = "Span trace";
+    public static final String SPAN_PARENT_KEY = "Span parent";
     public static final String PROJECT_AND_WORKSPACE_NAME_MISMATCH = "Project name and workspace name do not match the existing span";
 
     private final @NonNull SpanDAO spanDAO;
@@ -159,6 +161,7 @@ public Mono<UUID> create(@NonNull Span span) {
         var projectName = WorkspaceUtils.getProjectName(span.projectName());
         return idGenerator
                 .validateIdAsync(id, SPAN_KEY)
+                .then(Mono.fromRunnable(() -> validateSpanReferences(span.traceId(), span.parentSpanId())))
                 .then(projectService.getOrCreate(projectName))
                 .flatMap(project -> lockService.executeWithLock(
                         new LockService.Lock(id, SPAN_KEY),
@@ -220,7 +223,9 @@ public Mono<Void> update(@NonNull UUID id, @NonNull SpanUpdate spanUpdate) {
             String userName = ctx.get(RequestContext.USER_NAME);
 
             return idGenerator
-                    .validateIdForUpdateAsync(id, SPAN_KEY)
+                    .validateIdNotInFutureAsync(id, SPAN_KEY)
+                    .then(Mono.fromRunnable(
+                            () -> validateSpanReferences(spanUpdate.traceId(), spanUpdate.parentSpanId())))
                     .then(Mono.defer(() -> getProjectById(spanUpdate)
                             .switchIfEmpty(Mono.defer(() -> projectService.getOrCreate(projectName)))
                             .subscribeOn(Schedulers.boundedElastic()))
@@ -247,7 +252,10 @@ public Mono<Void> batchUpdate(@NonNull SpanBatchUpdate batchUpdate) {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
             String userName = ctx.get(RequestContext.USER_NAME);
 
-            return spanDAO.bulkUpdate(batchUpdate.ids(), batchUpdate.update(), mergeTags)
+            return Mono
+                    .fromRunnable(() -> validateSpanReferences(batchUpdate.update().traceId(),
+                            batchUpdate.update().parentSpanId()))
+                    .then(spanDAO.bulkUpdate(batchUpdate.ids(), batchUpdate.update(), mergeTags))
                     .onErrorResume(TagOperations::mapTagLimitError)
                     .doOnSuccess(__ -> {
                         log.info("Completed batch update for '{}' spans", batchUpdate.ids().size());
@@ -367,6 +375,15 @@ public Mono<Long> create(@NonNull SpanBatch batch) {
 
         List<Span> dedupedSpans = dedupSpans(batch.spans());
 
+        // Fail fast on invalid ids BEFORE any side effect below (auto-stripped attachment deletion, project
+        // creation), so a rejected batch never mutates state.
+        dedupedSpans.forEach(span -> {
+            if (span.id() != null) {
+                idGenerator.validateId(span.id(), SPAN_KEY);
+            }
+            validateSpanReferences(span.traceId(), span.parentSpanId());
+        });
+
         List<String> projectNames = dedupedSpans
                 .stream()
                 .map(Span::projectName)
@@ -438,6 +455,13 @@ private List<Span> dedupSpans(List<Span> initialSpans) {
         return result;
     }
 
+    // Shared span reference-id policy: the trace (required) and parent (optional) must be time-ordered
+    // UUIDv7, past allowed. Used by every span write path so the rules can't drift between them.
+    private void validateSpanReferences(UUID traceId, UUID parentSpanId) {
+        idGenerator.validateIdNotInFuture(traceId, SPAN_TRACE_KEY);
+        idGenerator.validateIdNotInFutureIfPresent(parentSpanId, SPAN_PARENT_KEY);
+    }
+
     private List<Span> bindSpanToProjectAndId(List<Span> spans, List<Project> projects) {
         Map<String, Project> projectPerName = projects.stream()
                 .collect(Collectors.toMap(
@@ -460,6 +484,7 @@ private List<Span> bindSpanToProjectAndId(List<Span> spans, List<Project> projec
 
                     UUID id = span.id() == null ? idGenerator.generateId() : span.id();
                     idGenerator.validateId(id, SPAN_KEY);
+                    // trace/parent references are validated up front in create(SpanBatch) before side effects.
 
                     return span.toBuilder().id(id).projectId(project.id()).build();
                 })

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

@@ -322,7 +322,7 @@ public Mono<Void> update(@NonNull TraceUpdate traceUpdate, @NonNull UUID id) {
         var projectName = WorkspaceUtils.getProjectName(traceUpdate.projectName());
 
         return Mono.deferContextual(ctx -> idGenerator
-                .validateIdForUpdateAsync(id, TRACE_KEY)
+                .validateIdNotInFutureAsync(id, TRACE_KEY)
                 .then(getProjectById(traceUpdate)
                         .switchIfEmpty(Mono.defer(() -> projectService.getOrCreate(projectName)))
                         .subscribeOn(Schedulers.boundedElastic())

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

@@ -9,6 +9,7 @@
 import com.comet.opik.api.attachment.EntityType;
 import com.comet.opik.api.attachment.StartMultipartUploadRequest;
 import com.comet.opik.api.attachment.StartMultipartUploadResponse;
+import com.comet.opik.domain.IdGenerator;
 import com.comet.opik.domain.ProjectService;
 import com.comet.opik.infrastructure.OpikConfiguration;
 import com.comet.opik.infrastructure.auth.RequestContext;
@@ -119,12 +120,14 @@ class AttachmentServiceImpl implements AttachmentService {
     private final @NonNull ProjectService projectService;
     private final @NonNull OpikConfiguration config;
     private final @NonNull Provider<RequestContext> requestContext;
+    private final @NonNull IdGenerator idGenerator;
     private static final Tika tika = new Tika();
     private static final int MAX_ATTACHMENTS_PER_ENTITY = 1_000;
 
     @Override
     public StartMultipartUploadResponse startMultiPartUpload(@NonNull StartMultipartUploadRequest startUploadRequest,
             @NonNull String workspaceId, @NonNull String userName) {
+        idGenerator.validateIdNotInFuture(startUploadRequest.entityId(), startUploadRequest.entityType().getValue());
         if (config.getS3Config().isMinIO()) {
             return prepareMinIOUploadResponse(startUploadRequest);
         }
@@ -149,6 +152,8 @@ public StartMultipartUploadResponse startMultiPartUpload(@NonNull StartMultipart
     public void completeMultiPartUpload(@NonNull CompleteMultipartUploadRequest completeUploadRequest,
             @NonNull String workspaceId,
             @NonNull String userName) {
+        idGenerator.validateIdNotInFuture(completeUploadRequest.entityId(),
+                completeUploadRequest.entityType().getValue());
         // In case of MinIO complete is not needed, file is uploaded directly via BE
         if (config.getS3Config().isMinIO()) {
             log.info("Skipping completeMultiPartUpload for MinIO");
@@ -190,6 +195,7 @@ public void uploadAttachment(@NonNull AttachmentInfo attachmentInfo, byte[] data
     @Override
     public void uploadAttachmentInternal(@NonNull AttachmentInfo attachmentInfo, byte[] data,
             @NonNull String workspaceId, @NonNull String userName) {
+        idGenerator.validateIdNotInFuture(attachmentInfo.entityId(), attachmentInfo.entityType().getValue());
 
         attachmentInfo = attachmentInfo.toBuilder()
                 .containerId(getProjectIdByName(attachmentInfo.projectName(), workspaceId, userName))