From 0c44a0c4a71ec5066fbe22832ae462cdc263c2ef Mon Sep 17 00:00:00 2001 From: Andrii Koropets Date: Mon, 17 Aug 2026 22:37:43 +0300 Subject: [PATCH] Render source in ScriptExecution --- .../rmm/CommandExecutionRepository.java | 2 +- .../rmm/CustomCommandExecutionRepository.java | 8 +++ .../CustomCommandExecutionRepositoryImpl.java | 35 ++++++++++++ .../rmm/CustomScriptExecutionRepository.java | 55 +------------------ .../CustomScriptExecutionRepositoryImpl.java | 18 ++++++ .../CommandExecutionStatusUpdateHandler.java | 2 +- .../ScriptExecutionStatusUpdateHandler.java | 2 +- ...mmandExecutionStatusUpdateHandlerTest.java | 14 ++--- ...criptExecutionStatusUpdateHandlerTest.java | 37 +++++++------ 9 files changed, 92 insertions(+), 81 deletions(-) create mode 100644 openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepository.java create mode 100644 openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepositoryImpl.java diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CommandExecutionRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CommandExecutionRepository.java index d80a6e3d5..1ff9b4649 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CommandExecutionRepository.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CommandExecutionRepository.java @@ -13,7 +13,7 @@ * {@code ScriptExecutionRepository}. */ @Repository -public interface CommandExecutionRepository extends MongoRepository { +public interface CommandExecutionRepository extends MongoRepository, CustomCommandExecutionRepository { /** Tenant-scoped lookup by Mongo {@code _id} — backs Relay {@code node(id)} refetch. */ Optional findByTenantIdAndId(String tenantId, String id); diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepository.java new file mode 100644 index 000000000..a2c1112bc --- /dev/null +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepository.java @@ -0,0 +1,8 @@ +package com.openframe.data.repository.rmm; + +import com.openframe.data.document.rmm.CommandExecution; + +public interface CustomCommandExecutionRepository { + + void applyResult(CommandExecution row); +} diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepositoryImpl.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepositoryImpl.java new file mode 100644 index 000000000..5842c8355 --- /dev/null +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomCommandExecutionRepositoryImpl.java @@ -0,0 +1,35 @@ +package com.openframe.data.repository.rmm; + +import com.openframe.data.document.rmm.CommandExecution; +import lombok.RequiredArgsConstructor; +import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.query.Criteria; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; +import org.springframework.stereotype.Repository; + +@Repository +@RequiredArgsConstructor +public class CustomCommandExecutionRepositoryImpl implements CustomCommandExecutionRepository { + + private static final String FIELD_ID = "_id"; + + private final MongoTemplate mongoTemplate; + + @Override + public void applyResult(CommandExecution row) { + Update update = new Update() + .set("status", row.getStatus()) + .set("statusChangedAt", row.getStatusChangedAt()) + .set("finishedAt", row.getFinishedAt()) + .set("exitCode", row.getExitCode()) + .set("executionTimeMs", row.getExecutionTimeMs()) + .set("timedOut", row.getTimedOut()) + .set("stdout", row.getStdout()) + .set("stdoutTruncated", row.getStdoutTruncated()) + .set("stderr", row.getStderr()) + .set("stderrTruncated", row.getStderrTruncated()) + .set("error", row.getError()); + mongoTemplate.updateFirst(new Query(Criteria.where(FIELD_ID).is(row.getId())), update, CommandExecution.class); + } +} diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepository.java index f05ef3726..47f9c84b8 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepository.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepository.java @@ -13,38 +13,11 @@ * Custom MongoTemplate-backed queries for {@link ScriptExecution}. Owner-scoped — * both per-script (Script → Execution History tab) and per-schedule (Schedule → * Execution History tab) queries share one API, differing only in {@link ExecutionOwnerScope}. - * - *

Spring Data derived methods cannot express id-cursor pagination + facet aggregations - * over shared predicate variations, so the list / count / facet queries live here rather - * than in the service — mirrors {@code CustomScriptRepository}. Implementation: - * {@code CustomScriptExecutionRepositoryImpl}. */ public interface CustomScriptExecutionRepository { - /** - * Cursor-paginated executions for one owner (script or schedule) within a tenant. - * - *

Cursor semantics: the cursor is the raw {@code ObjectId} hex from the boundary - * row of the previous page. With {@code backward=true} the scan walks in the - * opposite direction so paging "before" a cursor returns the rows immediately newer - * than it; the caller is expected to reverse the returned list for display. - * - *

Pass {@code limit + 1} from the caller to detect whether more pages exist - * beyond this one (the canonical "fetch one extra" trick). - * - * @param tenantId tenant scope — required, never null - * @param owner what narrows the base predicate — see {@link ExecutionOwnerScope} - * @param filter optional extra constraints (statuses/initiators/machines); null / empty - * fields impose no constraint - * @param sortField sort field, must satisfy {@link #isSortableField} - * @param sortDirection sort direction - * @param cursor raw {@code _id} cursor (already base64-decoded); null = first page - * @param backward true when paginating with {@code before/last} - * @param limit max rows to return (usually {@code pageSize + 1}) - * @param search optional case-insensitive substring matched across - * {@code executionId}, {@code machineId}, {@code stdout}, - * {@code stderr}; null/blank imposes no constraint - */ + void applyResult(ScriptExecution row); + List findPage(String tenantId, ExecutionOwnerScope owner, ScriptExecutionQueryFilter filter, @@ -55,45 +28,21 @@ List findPage(String tenantId, int limit, String search); - /** - * Full matching count for the {@code (tenantId, owner, filter, search)} tuple, ignoring - * pagination. Backs the connection's {@code filteredCount} so the UI can show the full - * total immediately while items load page by page. - */ long count(String tenantId, ExecutionOwnerScope owner, ScriptExecutionQueryFilter filter, String search); - /** - * Faceted options for one owner's Execution History: {@code value → matching count}. - * Applies every filter arm EXCEPT the facet's own field, so its dropdown keeps offering - * every switchable value. Labels are resolved by the service. - */ Map facet(String tenantId, ExecutionOwnerScope owner, ScriptExecutionQueryFilter filter, String search, ExecutionFacetField facet); - /** Whether the given field is allowed as a sort key. */ boolean isSortableField(String field); - /** Default sort field when none is supplied. */ String getDefaultSortField(); - /** - * Encode the compound-keyset cursor for a page boundary row: {@code |} - * for non-{@code _id} sort fields (empty {@code sortValue} for null Instant), or the plain - * hex {@code _id} for {@code _id} sort. Consumed by {@link #findPage} on the next request. - */ String encodeCursor(com.openframe.data.document.rmm.ScriptExecution row, String sortField); - /** - * Count leaf {@link ScriptExecution} rows for one schedule fire, grouped by status. - * A single {@code $match + $group} pass — one round-trip — backs the header aggregator - * so it can decide "any leaf still running? any failed?" without loading the rows - * themselves. Tenant-scoped so it hits the compound index. - */ LeafStatusCounts countLeavesByStatus(String tenantId, String executionId); - /** Running/failed counts for the leaves of one schedule fire; other terminal statuses are irrelevant to the decision. */ record LeafStatusCounts(long running, long failed) {} } diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepositoryImpl.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepositoryImpl.java index 60bf98252..139a178ed 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepositoryImpl.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptExecutionRepositoryImpl.java @@ -15,6 +15,7 @@ import org.springframework.data.mongodb.core.aggregation.AggregationResults; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; import org.springframework.stereotype.Repository; import java.time.Instant; @@ -83,6 +84,23 @@ public class CustomScriptExecutionRepositoryImpl implements CustomScriptExecutio private final MongoTemplate mongoTemplate; + @Override + public void applyResult(ScriptExecution row) { + Update update = new Update() + .set(FIELD_STATUS, row.getStatus()) + .set(FIELD_STATUS_CHANGED_AT, row.getStatusChangedAt()) + .set(FIELD_FINISHED_AT, row.getFinishedAt()) + .set("exitCode", row.getExitCode()) + .set("executionTimeMs", row.getExecutionTimeMs()) + .set("timedOut", row.getTimedOut()) + .set(FIELD_STDOUT, row.getStdout()) + .set("stdoutTruncated", row.getStdoutTruncated()) + .set(FIELD_STDERR, row.getStderr()) + .set("stderrTruncated", row.getStderrTruncated()) + .set("error", row.getError()); + mongoTemplate.updateFirst(new Query(Criteria.where(FIELD_ID).is(row.getId())), update, ScriptExecution.class); + } + @Override public List findPage(String tenantId, ExecutionOwnerScope owner, ScriptExecutionQueryFilter filter, diff --git a/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandler.java b/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandler.java index 369617106..e28985aeb 100644 --- a/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandler.java +++ b/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandler.java @@ -110,7 +110,7 @@ private void applyResult(CommandExecution row, JsonNode after) { row.setStderr(truncStderr.value); row.setStderrTruncated(truncStderr.truncated); - commandExecutionRepository.save(row); + commandExecutionRepository.applyResult(row); log.info("Transitioned CommandExecution row: executionId={} machineId={} status=RUNNING→{} exitCode={} timedOut={}", row.getExecutionId(), row.getMachineId(), newStatus, exitCode, timedOut); } diff --git a/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandler.java b/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandler.java index c38001cf5..bf392a8ca 100644 --- a/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandler.java +++ b/openframe-stream-service-core/src/main/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandler.java @@ -168,7 +168,7 @@ private void applyResult(ScriptExecution row, JsonNode after) { row.setStderr(truncStderr.value); row.setStderrTruncated(truncStderr.truncated); - scriptExecutionRepository.save(row); + scriptExecutionRepository.applyResult(row); log.info("Transitioned Execution row: executionId={} status=RUNNING→{} exitCode={} timedOut={}", row.getExecutionId(), newStatus, exitCode, timedOut); } diff --git a/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandlerTest.java b/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandlerTest.java index a98bfe9e9..4e2253b45 100644 --- a/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandlerTest.java +++ b/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/CommandExecutionStatusUpdateHandlerTest.java @@ -65,7 +65,7 @@ void handle_success_transitionsRowToSuccess() { handler.handle(messageWith(0, false, null, 42L, "ok\n", ""), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(CommandExecution.class); - verify(commandExecutionRepository).save(captor.capture()); + verify(commandExecutionRepository).applyResult(captor.capture()); CommandExecution saved = captor.getValue(); assertThat(saved.getStatus()).isEqualTo(ExecutionStatus.SUCCESS); assertThat(saved.getExitCode()).isZero(); @@ -87,7 +87,7 @@ void handle_nonZeroExit_transitionsRowToFailing() { handler.handle(messageWith(1, false, null, null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(CommandExecution.class); - verify(commandExecutionRepository).save(captor.capture()); + verify(commandExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); } @@ -101,7 +101,7 @@ void handle_timedOut_transitionsRowToFailing() { handler.handle(messageWith(null, true, null, null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(CommandExecution.class); - verify(commandExecutionRepository).save(captor.capture()); + verify(commandExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); assertThat(captor.getValue().getTimedOut()).isTrue(); } @@ -116,7 +116,7 @@ void handle_agentError_transitionsRowToFailing() { handler.handle(messageWith(0, false, "SHELL_UNAVAILABLE", null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(CommandExecution.class); - verify(commandExecutionRepository).save(captor.capture()); + verify(commandExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); assertThat(captor.getValue().getError()).isEqualTo("SHELL_UNAVAILABLE"); } @@ -131,7 +131,7 @@ void handle_alreadyTerminal_doesNotOverwrite() { handler.handle(messageWith(0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(commandExecutionRepository, never()).save(any()); + verify(commandExecutionRepository, never()).applyResult(any()); } @Test @@ -142,7 +142,7 @@ void handle_rowMissing_skipsSaveQuietly() { handler.handle(messageWith(0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(commandExecutionRepository, never()).save(any()); + verify(commandExecutionRepository, never()).applyResult(any()); } @Test @@ -156,7 +156,7 @@ void handle_truncatesLargeStdoutAndStderr() { handler.handle(messageWith(0, false, null, null, huge, huge), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(CommandExecution.class); - verify(commandExecutionRepository).save(captor.capture()); + verify(commandExecutionRepository).applyResult(captor.capture()); CommandExecution saved = captor.getValue(); assertThat(saved.getStdout().getBytes(StandardCharsets.UTF_8).length) .isLessThanOrEqualTo(CommandExecution.MAX_OUTPUT_BYTES); diff --git a/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandlerTest.java b/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandlerTest.java index f43f019c4..d62fffa65 100644 --- a/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandlerTest.java +++ b/openframe-stream-service-core/src/test/java/com/openframe/stream/handler/ScriptExecutionStatusUpdateHandlerTest.java @@ -77,7 +77,7 @@ void handle_success_transitionsRowToSuccess() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, 42L, "ok\n", ""), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); ScriptExecution saved = captor.getValue(); assertThat(saved.getStatus()).isEqualTo(ExecutionStatus.SUCCESS); assertThat(saved.getExitCode()).isZero(); @@ -156,8 +156,8 @@ void handle_sentinelNotDispatched_doesNotSave() { } @Test - @DisplayName("handle: SCHEDULED source survives the RUNNING→SUCCESS transition (regression: source must not be dropped on completion)") - void handle_success_preservesScheduledSource() { + @DisplayName("handle: completion goes through the targeted applyResult ($set), NEVER a full-document save — the guard that stops a skewed deploy resetting source→MANUAL") + void handle_success_usesTargetedUpdateNotFullSave() { ScriptExecution row = runningRow(EXECUTION_ID); row.setSource(ExecutionSource.SCHEDULED); row.setScheduleId("sched-99"); @@ -167,9 +167,10 @@ void handle_success_preservesScheduledSource() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, 42L, "ok\n", ""), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.SUCCESS); - assertThat(captor.getValue().getSource()).isEqualTo(ExecutionSource.SCHEDULED); + // A full-document replace is what dropped `source`; the write-back must never use it. + verify(scriptExecutionRepository, never()).save(any()); } @Test @@ -182,7 +183,7 @@ void handle_nonZeroExit_transitionsRowToFailing() { handler.handle(messageWith(EXECUTION_ID, 1, false, null, null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); } @@ -196,7 +197,7 @@ void handle_timedOut_transitionsRowToFailing() { handler.handle(messageWith(EXECUTION_ID, null, true, null, null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); assertThat(captor.getValue().getTimedOut()).isTrue(); } @@ -211,7 +212,7 @@ void handle_agentError_transitionsRowToFailing() { handler.handle(messageWith(EXECUTION_ID, 0, false, "SHELL_UNAVAILABLE", null, null, null), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.FAILED); assertThat(captor.getValue().getError()).isEqualTo("SHELL_UNAVAILABLE"); } @@ -226,7 +227,7 @@ void handle_alreadyTerminal_doesNotOverwrite() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository, never()).save(any()); + verify(scriptExecutionRepository, never()).applyResult(any()); } @Test @@ -237,7 +238,7 @@ void handle_rowMissing_skipsSaveQuietly() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository, never()).save(any()); + verify(scriptExecutionRepository, never()).applyResult(any()); } @Test @@ -251,7 +252,7 @@ void handle_truncatesLargeStdoutAndStderr() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, null, huge, huge), new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); ScriptExecution saved = captor.getValue(); assertThat(saved.getStdout().getBytes(StandardCharsets.UTF_8).length) .isLessThanOrEqualTo(ScriptExecution.MAX_OUTPUT_BYTES); @@ -297,7 +298,7 @@ void handle_noTenantId_stillMatchesAndTransitions() { handler.handle(message, new IntegratedToolEnrichedData()); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); - verify(scriptExecutionRepository).save(captor.capture()); + verify(scriptExecutionRepository).applyResult(captor.capture()); assertThat(captor.getValue().getStatus()).isEqualTo(ExecutionStatus.SUCCESS); } @@ -331,7 +332,7 @@ void handle_withScriptId_correlatesOnScriptId() { verify(scriptExecutionRepository).findByMachineIdAndExecutionIdAndScriptId(MACHINE_ID, EXECUTION_ID, "script-b"); // Must NOT fall back to the ambiguous two-field lookup when scriptId is present. verify(scriptExecutionRepository, never()).findByMachineIdAndExecutionId(any(), any()); - verify(scriptExecutionRepository).save(any(ScriptExecution.class)); + verify(scriptExecutionRepository).applyResult(any(ScriptExecution.class)); } @Test @@ -344,7 +345,7 @@ void handle_adHocRow_skipsAggregator() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, 5L, "ok", ""), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository).save(any(ScriptExecution.class)); + verify(scriptExecutionRepository).applyResult(any(ScriptExecution.class)); verifyNoInteractions(scheduleScriptExecutionAggregator); } @@ -358,7 +359,7 @@ void handle_scheduleRow_invokesAggregator() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, 5L, "ok", ""), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository).save(any(ScriptExecution.class)); + verify(scriptExecutionRepository).applyResult(any(ScriptExecution.class)); // From the row, NOT the wire — the leaf's persisted (tenantId, executionId) is the source of truth. verify(scheduleScriptExecutionAggregator).aggregate(TENANT_ID, EXECUTION_ID); } @@ -374,7 +375,7 @@ void handle_scheduleRowAlreadyTerminal_stillAggregates() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository, never()).save(any()); // row already terminal — no leaf write + verify(scriptExecutionRepository, never()).applyResult(any()); // row already terminal — no leaf write // Aggregator IS called: it is idempotent (short-circuits on any RUNNING leaf, atomic // conditional-update on all-terminal), and this is our recovery from a prior aggregate() // that threw AFTER the leaf save succeeded on a previous delivery of the same Kafka msg. @@ -392,7 +393,7 @@ void handle_adHocRowAlreadyTerminal_noAggregator() { handler.handle(messageWith(EXECUTION_ID, 0, false, null, null, null, null), new IntegratedToolEnrichedData()); - verify(scriptExecutionRepository, never()).save(any()); + verify(scriptExecutionRepository, never()).applyResult(any()); verifyNoInteractions(scheduleScriptExecutionAggregator); } @@ -412,7 +413,7 @@ void handle_missingScriptId_skipsQuietly() { handler.handle(message, new IntegratedToolEnrichedData()); verify(scriptExecutionRepository, never()).findByMachineIdAndExecutionIdAndScriptId(any(), any(), any()); - verify(scriptExecutionRepository, never()).save(any()); + verify(scriptExecutionRepository, never()).applyResult(any()); } private DeserializedDebeziumMessage messageWith(String executionId,