From 20af75e8769140ccce06d10b9f8c515d7119214f Mon Sep 17 00:00:00 2001 From: Thiago Hora Date: Sat, 5 Sep 2026 22:30:11 +0200 Subject: [PATCH] [OPIK-8262] [BE] fix: publish one stream entry per trace-thread id OnlineScorePublisher.enqueueThreadMessage built ONE stream entry carrying the whole thread-id list. BaseRedisSubscriber acks and removes per entry, so N thread ids under one entry share a single retry verdict: whichever outcome the consumer surfaces decides the fate of every sibling. A retry replays thread ids that already succeeded, and a drop discards ones that would have succeeded on the next attempt. Publishing one entry per thread id makes failure granularity match ack granularity. A retry then re-runs exactly the thread that failed, and no sibling is ever replayed. threadIds stays a @NotEmpty List on the message record even though this now puts a single id in it: a consumer running the new build must still read the multi-id entries the previous build left in the stream during a rolling upgrade, and narrowing the field would make those undecodable in exactly the deploy that ships this. Those entries are scored as a single message, as they are today; once publishing is fixed no new ones are created, so they are a finite draining population. An empty threadIds list now publishes nothing, where it previously published one entry whose threadIds violated the record's own @NotEmpty. Co-Authored-By: Claude Opus 5 (1M context) --- .../evaluators/OnlineScorePublisher.java | 50 +++-- .../TraceThreadOnlineScorerPublisher.java | 19 +- .../OnlineScorePublisherIntegrationTest.java | 204 ++++++++++++++++++ .../evaluators/OnlineScorePublisherTest.java | 133 +++++++++++- 4 files changed, 385 insertions(+), 21 deletions(-) create mode 100644 apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherIntegrationTest.java diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/OnlineScorePublisher.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/OnlineScorePublisher.java index ba7033362e4..d08d815e09b 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/OnlineScorePublisher.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/OnlineScorePublisher.java @@ -56,29 +56,29 @@ public interface OnlineScorePublisher { Mono enqueueMessage(List messages, AutomationRuleEvaluatorType type); /** - * Enqueues a thread message for scoring based on the provided rule. The returned publisher must be subscribed - * for the enqueue to happen. + * Enqueues thread messages for scoring — one stream entry per thread id, not one for the batch. + * The returned publisher must be subscribed for the enqueue to happen. * - * @param threadIds the IDs of the threads to score + * @param threadIds the IDs of the threads to score; one message is published per element * @param ruleId the ID of the rule to apply * @param projectId the ID of the project * @param workspaceId the ID of the workspace * @param userName the name of the user who initiated the scoring - * @return a {@link Mono} that completes once the message is enqueued + * @return a {@link Mono} that completes once all messages are enqueued */ Mono enqueueThreadMessage(List threadIds, UUID ruleId, UUID projectId, String workspaceId, String userName); /** - * Enqueues a thread message for an already-resolved rule, avoiding the blocking rule lookup that the - * {@code ruleId} overload performs. Prefer this when the caller already holds the {@link AutomationRuleEvaluator}. + * Enqueues thread messages for an already-resolved rule, avoiding the blocking rule lookup that the + * {@code ruleId} overload performs. Publishes one stream entry per thread id. * - * @param threadIds the IDs of the threads to score + * @param threadIds the IDs of the threads to score; one message is published per element * @param rule the already-resolved automation rule evaluator * @param projectId the ID of the project * @param workspaceId the ID of the workspace * @param userName the name of the user who initiated the scoring - * @return a {@link Mono} that completes once the message is enqueued + * @return a {@link Mono} that completes once all messages are enqueued */ Mono enqueueThreadMessage(List threadIds, AutomationRuleEvaluator rule, UUID projectId, String workspaceId, String userName); @@ -185,16 +185,27 @@ public Mono enqueueThreadMessage(@NonNull List threadIds, @NonNull AutomationRuleEvaluator rule, @NonNull UUID projectId, @NonNull String workspaceId, @NonNull String userName) { - // Caller already holds the resolved rule — no findById needed. + // Caller already holds the resolved rule -- no findById needed. + // + // One message PER THREAD ID, not one carrying the whole list: the subscriber acks and removes per + // stream entry, so an entry holding N ids forces N independent outcomes through a single verdict. + // Splitting means a retry replays exactly the thread that failed. Costs N entries where there was + // one; the streams are capped by streamMaxLen and trimmed non-strictly. return switch (rule) { case AutomationRuleEvaluatorTraceThreadLlmAsJudge llmAsJudge -> enqueueMessage( - List.of(toLlmAsJudgeMessage(threadIds, rule.getId(), projectId, workspaceId, userName, - llmAsJudge.getCode())), + threadIds.stream() + .map(threadId -> toLlmAsJudgeMessage(threadId, rule.getId(), projectId, workspaceId, + userName, llmAsJudge.getCode())) + .toList(), rule.getType()); case AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython definedMetricPython -> { if (serviceTogglesConfig.isTraceThreadPythonEvaluatorEnabled()) { - yield enqueueMessage(List.of(toDefinedMetricPython(threadIds, rule.getId(), projectId, - workspaceId, userName, definedMetricPython.getCode())), rule.getType()); + yield enqueueMessage( + threadIds.stream() + .map(threadId -> toDefinedMetricPython(threadId, rule.getId(), projectId, + workspaceId, userName, definedMetricPython.getCode())) + .toList(), + rule.getType()); } log.warn("Trace Thread online scoring python evaluator is disabled, skipping enqueueing " + "for ruleId: '{}'", rule.getId()); @@ -204,10 +215,14 @@ yield enqueueMessage(List.of(toDefinedMetricPython(threadIds, rule.getId(), proj }; } - private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(List threadIds, UUID ruleId, UUID projectId, + /** + * {@code threadIds} stays a list even though this puts one id in it: narrowing the field would make + * the multi-id entries left by the previous build undecodable during the rolling upgrade. + */ + private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(String threadId, UUID ruleId, UUID projectId, String workspaceId, String userName, TraceThreadLlmAsJudgeCode code) { return TraceThreadToScoreLlmAsJudge.builder() - .threadIds(threadIds) + .threadIds(List.of(threadId)) .ruleId(ruleId) .projectId(projectId) .workspaceId(workspaceId) @@ -216,10 +231,11 @@ private TraceThreadToScoreLlmAsJudge toLlmAsJudgeMessage(List threadIds, .build(); } - private TraceThreadToScoreUserDefinedMetricPython toDefinedMetricPython(List threadIds, UUID ruleId, + /** @see #toLlmAsJudgeMessage on why {@code threadIds} stays a list. */ + private TraceThreadToScoreUserDefinedMetricPython toDefinedMetricPython(String threadId, UUID ruleId, UUID projectId, String workspaceId, String userName, TraceThreadUserDefinedMetricPythonCode code) { return TraceThreadToScoreUserDefinedMetricPython.builder() - .threadIds(threadIds) + .threadIds(List.of(threadId)) .ruleId(ruleId) .projectId(projectId) .workspaceId(workspaceId) diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/threads/TraceThreadOnlineScorerPublisher.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/threads/TraceThreadOnlineScorerPublisher.java index 88c6d6f962b..d4312f67946 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/threads/TraceThreadOnlineScorerPublisher.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/threads/TraceThreadOnlineScorerPublisher.java @@ -10,6 +10,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Set; @@ -21,6 +22,8 @@ @RequiredArgsConstructor(onConstructor_ = @Inject) class TraceThreadOnlineScorerPublisher { + private static final int THREAD_ID_LOG_SAMPLE = 10; + private final @NonNull OnlineScorePublisher onlineScorePublisher; public Mono publish(@NonNull UUID projectId, @NonNull List closeThreads) { @@ -50,19 +53,31 @@ public Mono publish(@NonNull UUID projectId, @NonNull List log.info( "Enqueued threads: '{}' trace threads for ruleId: '{}' in projectId '{}' for workspaceId '{}'", - threadIds, ruleId, projectId, workspaceId)); + sampleOf(threadIds), ruleId, projectId, workspaceId)); }) .then(); }); } + /** + * Bounded so an arbitrarily large close batch can't dump its whole id list into the logs. Substituted into + * the existing message unchanged — the template and its parameter order are what dashboards group on. + */ + private static String sampleOf(Collection threadIds) { + if (threadIds.size() <= THREAD_ID_LOG_SAMPLE) { + return threadIds.toString(); + } + return "%s (and %d more)".formatted( + threadIds.stream().limit(THREAD_ID_LOG_SAMPLE).toList(), threadIds.size() - THREAD_ID_LOG_SAMPLE); + } + private boolean isSampled(Map.Entry entry) { // Check if the entry is sampled (true) or not (false) return entry.getValue(); diff --git a/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherIntegrationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherIntegrationTest.java new file mode 100644 index 00000000000..0f881af4194 --- /dev/null +++ b/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherIntegrationTest.java @@ -0,0 +1,204 @@ +package com.comet.opik.domain.evaluators; + +import com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadLlmAsJudge; +import com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython; +import com.comet.opik.api.evaluators.AutomationRuleEvaluatorType; +import com.comet.opik.api.events.TraceThreadToScoreLlmAsJudge; +import com.comet.opik.api.events.TraceThreadToScoreUserDefinedMetricPython; +import com.comet.opik.api.resources.utils.RedisContainerUtils; +import com.comet.opik.infrastructure.OnlineScoringConfig; +import com.comet.opik.infrastructure.ServiceTogglesConfig; +import com.comet.opik.infrastructure.redis.RedisStreamCodec; +import com.comet.opik.podam.PodamFactoryUtils; +import com.redis.testcontainers.RedisContainer; +import org.apache.commons.lang3.RandomStringUtils; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; +import org.redisson.Redisson; +import org.redisson.api.RStreamReactive; +import org.redisson.api.RedissonReactiveClient; +import org.redisson.api.stream.StreamMessageId; +import org.redisson.config.Config; +import uk.co.jemos.podam.api.PodamFactory; + +import java.util.List; +import java.util.UUID; + +import static com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadLlmAsJudge.TraceThreadLlmAsJudgeCode; +import static com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython.TraceThreadUserDefinedMetricPythonCode; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * OPIK-8262, cross-layer half. The unit tests assert what the publisher is asked to write; this asserts what + * actually lands on a real Redis stream and survives the shipped codec — one entry per thread id, each decodable + * with its own id intact. + * + *

Stops at the stream rather than driving a scorer end to end: standing up the full scoring stack (ClickHouse + * traces, an LLM provider) to re-observe a Redis write would test the harness more than the change. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +class OnlineScorePublisherIntegrationTest { + + private final RedisContainer redis = RedisContainerUtils.newRedisContainer(); + private final PodamFactory podamFactory = PodamFactoryUtils.newPodamFactory(); + private final ServiceTogglesConfig serviceTogglesConfig = new ServiceTogglesConfig(); + + @Mock + private AutomationRuleEvaluatorService automationRuleEvaluatorService; + + private RedissonReactiveClient redissonClient; + private OnlineScoringConfig config; + private String streamName; + private String pythonStreamName; + + @BeforeAll + void setUpAll() { + redis.start(); + var redissonConfig = new Config(); + redissonConfig.useSingleServer().setAddress(redis.getRedisURI()).setDatabase(0); + redissonClient = Redisson.create(redissonConfig).reactive(); + + streamName = randomStreamName(); + pythonStreamName = randomStreamName(); + // The Python evaluator is behind a toggle; without it the publisher skips the enqueue entirely. + serviceTogglesConfig.setTraceThreadPythonEvaluatorEnabled(true); + config = OnlineScoringConfig.builder() + .streamMaxLen(10_000) + .streamTrimLimit(100) + .streams(List.of( + OnlineScoringConfig.StreamConfiguration.builder() + .scorer(AutomationRuleEvaluatorType.TRACE_THREAD_LLM_AS_JUDGE.getType()) + .streamName(streamName) + .codec(RedisStreamCodec.JAVA.getName()) + .build(), + OnlineScoringConfig.StreamConfiguration.builder() + .scorer(AutomationRuleEvaluatorType.TRACE_THREAD_USER_DEFINED_METRIC_PYTHON.getType()) + .streamName(pythonStreamName) + .codec(RedisStreamCodec.JAVA.getName()) + .build())) + .build(); + } + + // Same shape as RedisStreamCodecTest, the closest sibling: null-guarded shutdown then stop. + @AfterAll + void tearDownAll() { + if (redissonClient != null) { + redissonClient.shutdown(); + } + if (redis != null) { + redis.stop(); + } + } + + @Test + @DisplayName("A batch enqueue lands one decodable entry per thread id on the stream") + void enqueueThreadMessageWritesOneEntryPerThreadId() { + var publisher = newPublisher(); + var threadIds = List.of("thread-a-" + suffix(), "thread-b-" + suffix(), "thread-c-" + suffix()); + var rule = AutomationRuleEvaluatorTraceThreadLlmAsJudge.builder() + .id(UUID.randomUUID()) + .name(podamFactory.manufacturePojo(String.class)) + .code(podamFactory.manufacturePojo(TraceThreadLlmAsJudgeCode.class)) + .build(); + + var projectId = UUID.randomUUID(); + + publisher.enqueueThreadMessage(threadIds, rule, projectId, "workspace", "user").block(); + + // Compared whole, against independently built expectations: asserting only the thread ids would pass + // while the codec silently dropped the evaluator code, the user, or the project. Plain element + // equality is enough -- every type in the payload is a record, so generated equals covers each + // component and picks up any added later. + var expected = threadIds.stream() + .map(threadId -> TraceThreadToScoreLlmAsJudge.builder() + .threadIds(List.of(threadId)) + .ruleId(rule.getId()) + .projectId(projectId) + .workspaceId("workspace") + .userName("user") + .code(rule.getCode()) + .build()) + .toList(); + + var written = readStream(); + assertThat(written) + .as("the subscriber acks per entry, so each thread id needs its own entry to be retried alone") + .hasSize(threadIds.size()); + assertThat(written).containsExactlyInAnyOrderElementsOf(expected); + } + + /** + * The Python payload's mirror of the case above. Without it, the Python side is only ever asserted on the + * arguments handed to the publisher, before serialization — so a codec regression losing + * {@code code.metric} would leave the evaluator with nothing to run and no test would notice. This PR + * changes the fan-out for both payload types, so both need the same end-to-end check. + */ + @Test + @DisplayName("A Python batch enqueue lands one decodable entry per thread id on its own stream") + void enqueueThreadMessageWritesOnePythonEntryPerThreadId() { + var publisher = newPublisher(); + var threadIds = List.of("thread-a-" + suffix(), "thread-b-" + suffix(), "thread-c-" + suffix()); + var projectId = UUID.randomUUID(); + var rule = AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython.builder() + .id(UUID.randomUUID()) + .name(podamFactory.manufacturePojo(String.class)) + .code(podamFactory.manufacturePojo(TraceThreadUserDefinedMetricPythonCode.class)) + .build(); + + publisher.enqueueThreadMessage(threadIds, rule, projectId, "workspace", "user").block(); + + var expected = threadIds.stream() + .map(threadId -> TraceThreadToScoreUserDefinedMetricPython.builder() + .threadIds(List.of(threadId)) + .ruleId(rule.getId()) + .projectId(projectId) + .workspaceId("workspace") + .userName("user") + .code(rule.getCode()) + .build()) + .toList(); + + List written = readStream(pythonStreamName); + assertThat(written) + .as("the Python payload needs the same per-entry granularity as the LLM one") + .hasSize(threadIds.size()); + assertThat(written).containsExactlyInAnyOrderElementsOf(expected); + } + + private OnlineScorePublisher newPublisher() { + redissonClient.getStream(streamName, RedisStreamCodec.JAVA.getCodec()).delete().block(); + redissonClient.getStream(pythonStreamName, RedisStreamCodec.JAVA.getCodec()).delete().block(); + return new OnlineScorePublisherImpl(config, serviceTogglesConfig, redissonClient, + automationRuleEvaluatorService); + } + + private List readStream() { + return readStream(streamName); + } + + private List readStream(String name) { + RStreamReactive stream = redissonClient.getStream(name, RedisStreamCodec.JAVA.getCodec()); + return stream.range(StreamMessageId.MIN, StreamMessageId.MAX).block() + .values().stream() + .map(entry -> entry.get(OnlineScoringConfig.PAYLOAD_FIELD)) + .toList(); + } + + private static String randomStreamName() { + return "test-stream-%s".formatted(RandomStringUtils.secure().nextAlphanumeric(10).toLowerCase()); + } + + private static String suffix() { + return RandomStringUtils.secure().nextAlphanumeric(16); + } +} diff --git a/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherTest.java b/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherTest.java index 3b4edb16c4b..275111a994e 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/domain/evaluators/OnlineScorePublisherTest.java @@ -1,13 +1,21 @@ package com.comet.opik.domain.evaluators; +import com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadLlmAsJudge; +import com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython; import com.comet.opik.api.evaluators.AutomationRuleEvaluatorType; +import com.comet.opik.api.events.TraceThreadToScoreLlmAsJudge; +import com.comet.opik.api.events.TraceThreadToScoreUserDefinedMetricPython; import com.comet.opik.infrastructure.OnlineScoringConfig; import com.comet.opik.infrastructure.ServiceTogglesConfig; import com.comet.opik.infrastructure.redis.RedisStreamCodec; import com.comet.opik.podam.PodamFactoryUtils; import org.apache.commons.lang3.RandomStringUtils; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -19,12 +27,17 @@ import uk.co.jemos.podam.api.PodamFactory; import java.util.List; +import java.util.UUID; +import java.util.stream.IntStream; +import static com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadLlmAsJudge.TraceThreadLlmAsJudgeCode; +import static com.comet.opik.api.evaluators.AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython.TraceThreadUserDefinedMetricPythonCode; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -79,9 +92,116 @@ void shouldUsePerStreamStreamAddParams() { }); } + /** + * OPIK-8262. {@code enqueueThreadMessage} used to build ONE stream entry carrying the whole thread-id + * list. {@code BaseRedisSubscriber} acks and removes per entry, so N thread ids under one entry share a + * single retry verdict — and once {@code ChatCompletionService.scoreTrace} classifies provider errors by + * status (same change), a permanent 400 and a transient 429 can land under one entry and whichever the + * consumer surfaces mis-serves the other. Publishing one entry per thread id makes failure granularity + * match ack granularity, so a retry replays only the thread that failed. + */ + @Nested + @DisplayName("One stream entry per thread id") + class ThreadMessageFanOutTests { + + @ParameterizedTest(name = "llm-as-judge: {0} thread ids -> {0} entries, one id each") + @ValueSource(ints = {1, 2, 5}) + void shouldPublishOneLlmAsJudgeEntryPerThreadId(int threadCount) { + var config = createConfig(AutomationRuleEvaluatorType.TRACE_THREAD_LLM_AS_JUDGE); + var publisher = createPublisher(config); + var threadIds = threadIds(threadCount); + var rule = AutomationRuleEvaluatorTraceThreadLlmAsJudge.builder() + .id(UUID.randomUUID()) + .name(podamFactory.manufacturePojo(String.class)) + .code(podamFactory.manufacturePojo(TraceThreadLlmAsJudgeCode.class)) + .build(); + + publisher.enqueueThreadMessage(threadIds, rule, UUID.randomUUID(), + podamFactory.manufacturePojo(String.class), podamFactory.manufacturePojo(String.class)).block(); + + var published = capturePayloads(TraceThreadToScoreLlmAsJudge.class); + assertThat(published) + .as("one stream entry per thread id, so the subscriber's per-entry ack retires exactly one" + + " thread's work") + .hasSize(threadCount); + assertThat(published) + .allSatisfy(message -> assertThat(message.threadIds()) + .as("each entry carries exactly one thread id — the whole point of the split") + .hasSize(1)); + assertThat(published.stream().map(message -> message.threadIds().getFirst()).toList()) + .as("every requested thread id is still enqueued, none lost in the split") + .containsExactlyInAnyOrderElementsOf(threadIds); + } + + @ParameterizedTest(name = "python: {0} thread ids -> {0} entries, one id each") + @ValueSource(ints = {1, 2, 5}) + void shouldPublishOnePythonEntryPerThreadId(int threadCount) { + serviceTogglesConfig.setTraceThreadPythonEvaluatorEnabled(true); + var config = createConfig(AutomationRuleEvaluatorType.TRACE_THREAD_USER_DEFINED_METRIC_PYTHON); + var publisher = createPublisher(config); + var threadIds = threadIds(threadCount); + var rule = AutomationRuleEvaluatorTraceThreadUserDefinedMetricPython.builder() + .id(UUID.randomUUID()) + .name(podamFactory.manufacturePojo(String.class)) + .code(podamFactory.manufacturePojo(TraceThreadUserDefinedMetricPythonCode.class)) + .build(); + + publisher.enqueueThreadMessage(threadIds, rule, UUID.randomUUID(), + podamFactory.manufacturePojo(String.class), podamFactory.manufacturePojo(String.class)).block(); + + var published = capturePayloads(TraceThreadToScoreUserDefinedMetricPython.class); + assertThat(published).hasSize(threadCount); + assertThat(published).allSatisfy(message -> assertThat(message.threadIds()).hasSize(1)); + assertThat(published.stream().map(message -> message.threadIds().getFirst()).toList()) + .containsExactlyInAnyOrderElementsOf(threadIds); + } + + /** + * An empty list used to publish one entry whose {@code threadIds} violated the record's + * {@code @NotEmpty} — a message no consumer could do anything with. Zero ids now means zero entries. + */ + @Test + @DisplayName("An empty thread-id list publishes nothing at all") + void shouldPublishNothingForAnEmptyThreadIdList() { + var config = createConfig(AutomationRuleEvaluatorType.TRACE_THREAD_LLM_AS_JUDGE); + var publisher = createPublisherWithoutStreamStubs(config); + var rule = AutomationRuleEvaluatorTraceThreadLlmAsJudge.builder() + .id(UUID.randomUUID()) + .name(podamFactory.manufacturePojo(String.class)) + .code(podamFactory.manufacturePojo(TraceThreadLlmAsJudgeCode.class)) + .build(); + + publisher.enqueueThreadMessage(List.of(), rule, UUID.randomUUID(), + podamFactory.manufacturePojo(String.class), podamFactory.manufacturePojo(String.class)).block(); + + verify(stream, never()).add(any()); + } + + private List threadIds(int count) { + return IntStream.range(0, count) + .mapToObj(index -> "thread-%d-%s".formatted(index, + RandomStringUtils.secure().nextAlphanumeric(16))) + .toList(); + } + + private List capturePayloads(Class messageType) { + ArgumentCaptor> captor = ArgumentCaptor.forClass(StreamAddParams.class); + verify(stream, atLeastOnce()).add(captor.capture()); + return captor.getAllValues().stream() + .map(params -> params.getEntries().get(OnlineScoringConfig.PAYLOAD_FIELD)) + .map(messageType::cast) + .toList(); + } + } + private OnlineScorePublisher createPublisher(OnlineScoringConfig onlineScoringConfig) { - when(redisClient.getStream(anyString(), any())).thenReturn(stream); when(stream.add(any())).thenReturn(Mono.just(new StreamMessageId(System.currentTimeMillis(), 0))); + return createPublisherWithoutStreamStubs(onlineScoringConfig); + } + + /** For the cases that assert nothing is ever added — strict stubbing rejects an unused {@code add} stub. */ + private OnlineScorePublisher createPublisherWithoutStreamStubs(OnlineScoringConfig onlineScoringConfig) { + when(redisClient.getStream(anyString(), any())).thenReturn(stream); return new OnlineScorePublisherImpl( onlineScoringConfig, serviceTogglesConfig, redisClient, automationRuleEvaluatorService); } @@ -97,8 +217,17 @@ private OnlineScoringConfig createConfig() { } private OnlineScoringConfig createConfig(Integer perStreamMaxLen, Integer perStreamTrimLimit) { + return createConfig(AutomationRuleEvaluatorType.LLM_AS_JUDGE, perStreamMaxLen, perStreamTrimLimit); + } + + private OnlineScoringConfig createConfig(AutomationRuleEvaluatorType type) { + return createConfig(type, null, null); + } + + private OnlineScoringConfig createConfig(AutomationRuleEvaluatorType type, Integer perStreamMaxLen, + Integer perStreamTrimLimit) { var streamConfiguration = OnlineScoringConfig.StreamConfiguration.builder() - .scorer(AutomationRuleEvaluatorType.LLM_AS_JUDGE.getType()) + .scorer(type.getType()) .streamName("test-stream-%s".formatted( RandomStringUtils.secure().nextAlphanumeric(10).toLowerCase())) .codec(RedisStreamCodec.JAVA.getName())