From b0dc2552f6abfd5529e25ccb11dab70e080e6969 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 11:31:07 +0000 Subject: [PATCH 01/10] Add MemoryDistillerAgent and memory distillation on completed AgentProcess - Create MemoryDistillationResult record with stored/revoked memory lists - Create MemoryDistillerAgent with @Agent, @Action, @AchievesGoal annotations using ai().withDefaultLlm().createObject() for LLM-based memory distillation - Add maxMemories parameter to KafkaAgenticAggregateRuntime constructor - Integrate memory distillation in resumeNextAgentTask: when AgentProcess finishes with COMPLETED status, run MemoryDistillerAgent to completion using run() and translate results into MemoryStoredEvent/MemoryRevokedEvent - Enforce net memory limit: stored - revoked <= maxMemories - currentCount - Handle gracefully: missing agent, null result, exceptions - Add comprehensive tests for distillation logic and limit enforcement Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/b0b7907c-8588-4b46-af3a-67b710ba24ad Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../beans/AgenticAggregateRuntimeFactory.java | 3 +- .../embabel/MemoryDistillationResult.java | 70 +++ .../agentic/embabel/MemoryDistillerAgent.java | 151 +++++ .../runtime/KafkaAgenticAggregateRuntime.java | 172 ++++- .../embabel/MemoryDistillationResultTest.java | 76 +++ .../AgenticAggregateAutoCreateTest.java | 6 +- .../KafkaAgenticAggregateRuntimeTest.java | 14 +- .../runtime/MemoryDistillationTest.java | 594 ++++++++++++++++++ .../runtime/ResumeNextAgentTaskTest.java | 6 +- 9 files changed, 1074 insertions(+), 18 deletions(-) create mode 100644 main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java create mode 100644 main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java create mode 100644 main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java create mode 100644 main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java index 0dba4f4c..1361ab2e 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java @@ -134,7 +134,8 @@ public AgenticAggregateRuntime getObject() { KafkaAggregateRuntime kafkaRuntime = createRuntime(agenticInfo, aggregate, agentPlatform); return new KafkaAgenticAggregateRuntime( - kafkaRuntime, objectMapper, agenticInfo.stateClass(), agentPlatform, aggregate); + kafkaRuntime, objectMapper, agenticInfo.stateClass(), agentPlatform, aggregate, + agenticInfo.maxMemories()); } @Override diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java new file mode 100644 index 00000000..a50b53a5 --- /dev/null +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java @@ -0,0 +1,70 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.elasticsoftware.akces.agentic.embabel; + +import java.util.List; + +/** + * Result object produced by the {@link MemoryDistillerAgent} after distilling + * relevant memories from a completed {@link com.embabel.agent.core.AgentProcess}. + * + *

Contains two lists: + *

+ * + * @param stored the list of memories to store; each entry contains subject, fact, + * citations, and reason fields + * @param revoked the list of memories to revoke; each entry contains memoryId and reason + */ +public record MemoryDistillationResult( + List stored, + List revoked +) { + + /** + * A memory entry to be stored. + * + * @param subject a short (1–2 word) topic label + * @param fact the fact to remember (max 200 characters) + * @param citations the source citation for the fact + * @param reason why this memory should be stored + */ + public record StoredMemory( + String subject, + String fact, + String citations, + String reason + ) { + } + + /** + * A memory entry to be revoked. + * + * @param memoryId the UUID of the memory to revoke + * @param reason why this memory should be revoked + */ + public record RevokedMemory( + String memoryId, + String reason + ) { + } +} diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java new file mode 100644 index 00000000..91dcf951 --- /dev/null +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -0,0 +1,151 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.elasticsoftware.akces.agentic.embabel; + +import com.embabel.agent.api.annotation.AchievesGoal; +import com.embabel.agent.api.annotation.Action; +import com.embabel.agent.api.annotation.Agent; +import com.embabel.agent.api.common.ActionContext; +import com.embabel.agent.api.common.PlannerType; +import com.embabel.agent.core.ActionInvocation; + +import java.util.List; + +/** + * Embabel agent responsible for distilling relevant memories from a successfully + * completed {@link com.embabel.agent.core.AgentProcess}. + * + *

When an agent process finishes with status {@code COMPLETED}, this agent is + * invoked with the process's execution history and all objects from its blackboard. + * It uses an LLM to analyze the process results and determine which memories should + * be stored and which existing memories should be revoked. + * + *

The agent produces a {@link MemoryDistillationResult} containing: + *

+ * + *

The net number of new memories (stored minus revoked) is constrained by the + * {@code maxNewMemories} binding on the blackboard, which is derived from the + * aggregate's {@code maxMemories} configuration minus the current memory count. + */ +@Agent(name = MemoryDistillerAgent.AGENT_NAME, + description = "Distills relevant memories from a completed agent process", + planner = PlannerType.UTILITY) +public class MemoryDistillerAgent { + + /** + * The well-known agent name used to locate this agent on the + * {@link com.embabel.agent.core.AgentPlatform}. + */ + public static final String AGENT_NAME = "MemoryDistillerAgent"; + + /** + * Distills memories from a completed agent process by analyzing the process + * history and blackboard contents using an LLM. + * + *

The method reads the following bindings from the blackboard: + *

+ * + * @param context the action context providing access to AI capabilities and blackboard + * @return a {@link MemoryDistillationResult} with memories to store and revoke + */ + @Action(description = "Distill relevant memories from a completed agent process") + @AchievesGoal(description = "Distill and manage memories from completed agent processes") + public MemoryDistillationResult distillMemories(ActionContext context) { + List history = context.getProcessContext().getAgentProcess() + .getBlackboard().objectsOfType(ActionInvocation.class); + List blackboardObjects = (List) context.getProcessContext() + .getAgentProcess().getBlackboard().get("blackboardObjects"); + List existingMemories = (List) context.getProcessContext() + .getAgentProcess().getBlackboard().get("existingMemories"); + int maxNewMemories = (int) context.getProcessContext() + .getAgentProcess().getBlackboard().get("maxNewMemories"); + + String prompt = buildPrompt(history, blackboardObjects, existingMemories, maxNewMemories); + + return context.ai() + .withDefaultLlm() + .createObject(prompt, MemoryDistillationResult.class); + } + + private String buildPrompt(List history, + List blackboardObjects, + List existingMemories, + int maxNewMemories) { + StringBuilder sb = new StringBuilder(); + sb.append("You are a memory distillation agent. Analyze the following completed agent process "); + sb.append("and determine which facts should be stored as new memories and which existing "); + sb.append("memories should be revoked (because they are outdated, incorrect, or superseded).\n\n"); + + sb.append("CONSTRAINTS:\n"); + sb.append("- The net number of new memories (stored count minus revoked count) must be <= ") + .append(maxNewMemories).append("\n"); + sb.append("- Each stored memory must have: subject (1-2 words), fact (max 200 chars), "); + sb.append("citations (source reference), and reason (why it should be stored)\n"); + sb.append("- Each revoked memory must have: memoryId (from existing memories) and reason\n"); + sb.append("- Only store memories that are actionable, likely to remain relevant, and "); + sb.append("cannot always be inferred from limited context\n"); + sb.append("- Only revoke memories that are clearly outdated, incorrect, or superseded "); + sb.append("by new information\n\n"); + + sb.append("PROCESS HISTORY (actions executed):\n"); + if (history != null && !history.isEmpty()) { + for (Object invocation : history) { + sb.append("- ").append(invocation).append("\n"); + } + } else { + sb.append("(no actions recorded)\n"); + } + + sb.append("\nBLACKBOARD OBJECTS (process context and results):\n"); + if (blackboardObjects != null && !blackboardObjects.isEmpty()) { + for (Object obj : blackboardObjects) { + sb.append("- [").append(obj.getClass().getSimpleName()).append("] ") + .append(obj).append("\n"); + } + } else { + sb.append("(no objects)\n"); + } + + sb.append("\nEXISTING MEMORIES:\n"); + if (existingMemories != null && !existingMemories.isEmpty()) { + for (Object memory : existingMemories) { + sb.append("- ").append(memory).append("\n"); + } + } else { + sb.append("(no existing memories)\n"); + } + + sb.append("\nReturn a JSON object with 'stored' (list of new memories to store) "); + sb.append("and 'revoked' (list of existing memories to revoke). "); + sb.append("If no memories should be stored or revoked, return empty lists."); + + return sb.toString(); + } +} diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index 728ad474..cc7324ac 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -17,12 +17,16 @@ package org.elasticsoftware.akces.agentic.runtime; +import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; import com.embabel.agent.core.AgentProcessStatusCode; +import com.embabel.agent.core.ProcessOptions; import jakarta.annotation.Nullable; import org.apache.kafka.common.errors.SerializationException; import org.elasticsoftware.akces.agentic.AgenticAggregateRuntime; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationResult; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillerAgent; import org.elasticsoftware.akces.agentic.events.AgentTaskAssignedEvent; import org.elasticsoftware.akces.agentic.events.AgentTaskFinishedEvent; import org.elasticsoftware.akces.agentic.events.MemoryRevokedEvent; @@ -44,9 +48,7 @@ import java.io.IOException; import java.time.Instant; -import java.util.Collection; -import java.util.List; -import java.util.Objects; +import java.util.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -71,6 +73,7 @@ public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime { private final Class stateClass; private final AgentPlatform agentPlatform; private final AgenticAggregate aggregate; + private final int maxMemories; /** Round-robin counter for selecting the next agent task to resume. */ private final AtomicInteger nextTaskIndex = new AtomicInteger(0); @@ -86,17 +89,20 @@ public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime { * @param aggregate the agentic aggregate instance whose * {@link AgenticAggregate#getCreateDomainEvent()} method provides the * auto-create event + * @param maxMemories the maximum number of memories allowed for this aggregate */ public KafkaAgenticAggregateRuntime(KafkaAggregateRuntime delegate, ObjectMapper objectMapper, Class stateClass, AgentPlatform agentPlatform, - AgenticAggregate aggregate) { + AgenticAggregate aggregate, + int maxMemories) { this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); this.stateClass = Objects.requireNonNull(stateClass, "stateClass must not be null"); this.agentPlatform = Objects.requireNonNull(agentPlatform, "agentPlatform must not be null"); this.aggregate = Objects.requireNonNull(aggregate, "aggregate must not be null"); + this.maxMemories = maxMemories; } // ------------------------------------------------------------------------- @@ -511,8 +517,166 @@ public void resumeNextAgentTask(Consumer protocolRecordConsumer, statusCode, Instant.now()); tickEvents = Stream.concat(tickEvents, Stream.of(finishedEvent)); + + // Distill memories from successfully completed processes + if (statusCode == AgentProcessStatusCode.COMPLETED) { + List memoryEvents = distillMemories(agentProcess, state); + if (!memoryEvents.isEmpty()) { + tickEvents = Stream.concat(tickEvents, memoryEvents.stream()); + } + } } delegate.processDomainEvents(tickEvents, task.agentProcessId(), protocolRecordConsumer, stateRecordSupplier); } + + // ------------------------------------------------------------------------- + // Memory distillation + // ------------------------------------------------------------------------- + + /** + * Distills relevant memories from a successfully completed {@link AgentProcess} + * by running the {@link MemoryDistillerAgent} to completion. + * + *

Creates a separate agent process for the {@link MemoryDistillerAgent}, populates + * its blackboard with the completed process's history and blackboard objects, the + * current memories, and the maximum number of net new memories allowed, then runs + * the process to completion via {@link AgentProcess#run()}. + * + *

The net memory constraint ensures that + * {@code storedCount - revokedCount <= maxMemories - currentMemoryCount}, preventing + * the memory system from exceeding its configured capacity. + * + * @param completedProcess the agent process that has completed successfully + * @param state the current aggregate state + * @return a list of {@link MemoryStoredEvent} and {@link MemoryRevokedEvent} instances; + * may be empty if no memories need to be changed + */ + private List distillMemories(AgentProcess completedProcess, AggregateState state) { + Agent memoryDistillerAgent = resolveMemoryDistillerAgent(); + if (memoryDistillerAgent == null) { + logger.debug("MemoryDistillerAgent not deployed on the platform; skipping memory distillation"); + return List.of(); + } + + List currentMemories = state instanceof MemoryAwareState mas + ? mas.getMemories() + : List.of(); + int maxNewMemories = Math.max(0, maxMemories - currentMemories.size()); + + Map bindings = new LinkedHashMap<>(); + bindings.put("history", completedProcess.getHistory()); + bindings.put("blackboardObjects", completedProcess.getBlackboard().getObjects()); + bindings.put("existingMemories", currentMemories); + bindings.put("maxNewMemories", maxNewMemories); + + try { + AgentProcess distillerProcess = agentPlatform.createAgentProcess( + memoryDistillerAgent, ProcessOptions.DEFAULT, bindings); + distillerProcess.run(); + + MemoryDistillationResult result = distillerProcess.getBlackboard() + .last(MemoryDistillationResult.class); + + if (result == null) { + logger.debug("MemoryDistillerAgent produced no result for aggregate {}", getName()); + return List.of(); + } + + return translateDistillationResult(result, state.getAggregateId(), currentMemories, maxNewMemories); + } catch (Exception e) { + logger.warn("Memory distillation failed for aggregate {}; proceeding without memory updates", + getName(), e); + return List.of(); + } + } + + /** + * Resolves the {@link MemoryDistillerAgent} from the {@link AgentPlatform}. + * + * @return the resolved agent, or {@code null} if not deployed + */ + private Agent resolveMemoryDistillerAgent() { + for (Agent agent : agentPlatform.agents()) { + if (MemoryDistillerAgent.AGENT_NAME.equals(agent.getName())) { + return agent; + } + } + return null; + } + + /** + * Translates a {@link MemoryDistillationResult} into a list of domain events, + * enforcing the net memory limit. + * + *

The constraint {@code stored.size() - revoked.size() <= maxNewMemories} is + * enforced by truncating stored memories when the limit would be exceeded. + * + * @param result the distillation result from the agent + * @param aggregateId the aggregate identifier for the events + * @param currentMemories the current memories from the aggregate state + * @param maxNewMemories the maximum number of net new memories allowed + * @return an unmodifiable list of domain events + */ + private List translateDistillationResult(MemoryDistillationResult result, + String aggregateId, + List currentMemories, + int maxNewMemories) { + List events = new ArrayList<>(); + Instant now = Instant.now(); + + // Collect valid revocations first (only for memories that actually exist) + Set currentMemoryIds = new HashSet<>(); + for (AgenticAggregateMemory mem : currentMemories) { + currentMemoryIds.add(mem.memoryId()); + } + + List validRevocations = new ArrayList<>(); + if (result.revoked() != null) { + for (MemoryDistillationResult.RevokedMemory revoked : result.revoked()) { + if (revoked.memoryId() != null && currentMemoryIds.contains(revoked.memoryId())) { + validRevocations.add(revoked); + } else { + logger.debug("Skipping revocation of non-existent memory '{}' for aggregate {}", + revoked != null ? revoked.memoryId() : "null", getName()); + } + } + } + + // Calculate how many stored memories we can accept + int revokedCount = validRevocations.size(); + int maxStoredCount = maxNewMemories + revokedCount; + + // Add revocation events + for (MemoryDistillationResult.RevokedMemory revoked : validRevocations) { + events.add(new MemoryRevokedEvent(aggregateId, revoked.memoryId(), revoked.reason(), now)); + } + + // Add stored events (respecting the limit) + if (result.stored() != null) { + int storedCount = 0; + for (MemoryDistillationResult.StoredMemory stored : result.stored()) { + if (storedCount >= maxStoredCount) { + logger.debug("Memory distillation limit reached ({} stored, {} revoked, max {}); " + + "truncating remaining stored memories for aggregate {}", + storedCount, revokedCount, maxNewMemories, getName()); + break; + } + events.add(new MemoryStoredEvent( + aggregateId, + UUID.randomUUID().toString(), + stored.subject(), + stored.fact(), + stored.citations(), + stored.reason(), + now)); + storedCount++; + } + } + + logger.debug("Memory distillation for aggregate {} produced {} stored and {} revoked events", + getName(), events.size() - validRevocations.size(), validRevocations.size()); + + return List.copyOf(events); + } } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java new file mode 100644 index 00000000..aa4d1eea --- /dev/null +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java @@ -0,0 +1,76 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.elasticsoftware.akces.agentic.embabel; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link MemoryDistillationResult}. + */ +class MemoryDistillationResultTest { + + @Test + void shouldCreateWithStoredAndRevokedMemories() { + var stored = List.of( + new MemoryDistillationResult.StoredMemory("testing", "Use JUnit 5", "src/test", "best practice"), + new MemoryDistillationResult.StoredMemory("logging", "Use SLF4J", "src/main", "consistency") + ); + var revoked = List.of( + new MemoryDistillationResult.RevokedMemory("mem-old-1", "superseded by new info") + ); + + var result = new MemoryDistillationResult(stored, revoked); + + assertThat(result.stored()).hasSize(2); + assertThat(result.revoked()).hasSize(1); + assertThat(result.stored().getFirst().subject()).isEqualTo("testing"); + assertThat(result.stored().getFirst().fact()).isEqualTo("Use JUnit 5"); + assertThat(result.revoked().getFirst().memoryId()).isEqualTo("mem-old-1"); + } + + @Test + void shouldCreateWithEmptyLists() { + var result = new MemoryDistillationResult(List.of(), List.of()); + + assertThat(result.stored()).isEmpty(); + assertThat(result.revoked()).isEmpty(); + } + + @Test + void storedMemoryShouldExposeAllFields() { + var stored = new MemoryDistillationResult.StoredMemory( + "conventions", "Use records for DTOs", "src/main/Model.java:10", "immutability"); + + assertThat(stored.subject()).isEqualTo("conventions"); + assertThat(stored.fact()).isEqualTo("Use records for DTOs"); + assertThat(stored.citations()).isEqualTo("src/main/Model.java:10"); + assertThat(stored.reason()).isEqualTo("immutability"); + } + + @Test + void revokedMemoryShouldExposeAllFields() { + var revoked = new MemoryDistillationResult.RevokedMemory("mem-123", "outdated"); + + assertThat(revoked.memoryId()).isEqualTo("mem-123"); + assertThat(revoked.reason()).isEqualTo("outdated"); + } +} diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java index 2baa6ab4..8b14b8e4 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java @@ -240,7 +240,7 @@ void initializeStateShouldCallGetCreateDomainEventAndDelegateToRuntime() throws ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, testBot); + delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100); when(delegate.getName()).thenReturn("TestBot"); @@ -275,7 +275,7 @@ public Class getStateClass() { ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, nullReturningAggregate); + delegate, objectMapper, TestBotState.class, agentPlatform, nullReturningAggregate, 100); assertThatNullPointerException() .isThrownBy(() -> runtime.initializeState(pr -> {}, (der, ip) -> {})) @@ -289,7 +289,7 @@ void initializeStateShouldPassConsumersThroughToDelegate() throws IOException { ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, testBot); + delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100); when(delegate.getName()).thenReturn("TestBot"); diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java index eb6cf968..8f94e5a4 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java @@ -108,7 +108,7 @@ public String getAggregateId() { @BeforeEach void setUp() { objectMapper = JsonMapper.builder().build(); - runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TestMemoryState.class, agentPlatform, aggregate); + runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100); } // ------------------------------------------------------------------------- @@ -145,7 +145,7 @@ void getMemoriesShouldReturnMemoriesFromMemoryAwareState() throws IOException { @Test void getMemoriesShouldReturnEmptyListForNonMemoryAwareState() throws IOException { - var plainRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, PlainState.class, agentPlatform, aggregate); + var plainRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, PlainState.class, agentPlatform, aggregate, 100); var state = new PlainState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); @@ -235,7 +235,7 @@ void getAgentPlatformShouldReturnInjectedPlatform() { void constructorShouldRejectNullDelegate() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - null, objectMapper, TestMemoryState.class, agentPlatform, aggregate)) + null, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100)) .withMessageContaining("delegate"); } @@ -243,7 +243,7 @@ void constructorShouldRejectNullDelegate() { void constructorShouldRejectNullObjectMapper() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, null, TestMemoryState.class, agentPlatform, aggregate)) + delegate, null, TestMemoryState.class, agentPlatform, aggregate, 100)) .withMessageContaining("objectMapper"); } @@ -251,7 +251,7 @@ void constructorShouldRejectNullObjectMapper() { void constructorShouldRejectNullStateClass() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, null, agentPlatform, aggregate)) + delegate, objectMapper, null, agentPlatform, aggregate, 100)) .withMessageContaining("stateClass"); } @@ -259,7 +259,7 @@ void constructorShouldRejectNullStateClass() { void constructorShouldRejectNullAgentPlatform() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestMemoryState.class, null, aggregate)) + delegate, objectMapper, TestMemoryState.class, null, aggregate, 100)) .withMessageContaining("agentPlatform"); } @@ -267,7 +267,7 @@ void constructorShouldRejectNullAgentPlatform() { void constructorShouldRejectNullAggregate() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestMemoryState.class, agentPlatform, null)) + delegate, objectMapper, TestMemoryState.class, agentPlatform, null, 100)) .withMessageContaining("aggregate"); } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java new file mode 100644 index 00000000..2ad98564 --- /dev/null +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java @@ -0,0 +1,594 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.elasticsoftware.akces.agentic.runtime; + +import com.embabel.agent.core.Agent; +import com.embabel.agent.core.AgentPlatform; +import com.embabel.agent.core.AgentProcess; +import com.embabel.agent.core.AgentProcessStatusCode; +import com.embabel.agent.core.Blackboard; +import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationResult; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillerAgent; +import org.elasticsoftware.akces.agentic.events.AgentTaskFinishedEvent; +import org.elasticsoftware.akces.agentic.events.MemoryRevokedEvent; +import org.elasticsoftware.akces.agentic.events.MemoryStoredEvent; +import org.elasticsoftware.akces.aggregate.*; +import org.elasticsoftware.akces.commands.CommandBus; +import org.elasticsoftware.akces.events.DomainEvent; +import org.elasticsoftware.akces.kafka.KafkaAggregateRuntime; +import org.elasticsoftware.akces.protocol.AggregateStateRecord; +import org.elasticsoftware.akces.protocol.PayloadEncoding; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.json.JsonMapper; + +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * Unit tests for the memory distillation integration in + * {@link KafkaAgenticAggregateRuntime#resumeNextAgentTask}. + */ +@ExtendWith(MockitoExtension.class) +class MemoryDistillationTest { + + /** State that implements both TaskAwareState and MemoryAwareState. */ + record MemoryTaskState( + String id, + List assignedTasks, + List memories + ) implements AggregateState, TaskAwareState, MemoryAwareState { + + @Override + public String getAggregateId() { + return id; + } + + @Override + public List getAssignedTasks() { + return assignedTasks; + } + + @Override + public TaskAwareState withAssignedTask(AssignedTask task) { + var newTasks = new ArrayList<>(assignedTasks); + newTasks.add(task); + return new MemoryTaskState(id, List.copyOf(newTasks), memories); + } + + @Override + public TaskAwareState withoutAssignedTask(String agentProcessId) { + return new MemoryTaskState(id, assignedTasks.stream() + .filter(t -> !t.agentProcessId().equals(agentProcessId)) + .toList(), memories); + } + + @Override + public List getMemories() { + return memories; + } + + @Override + public MemoryAwareState withMemory(AgenticAggregateMemory memory) { + var updated = new ArrayList<>(memories); + updated.add(memory); + return new MemoryTaskState(id, assignedTasks, List.copyOf(updated)); + } + + @Override + public MemoryAwareState withoutMemory(String memoryId) { + return new MemoryTaskState(id, assignedTasks, memories.stream() + .filter(m -> !m.memoryId().equals(memoryId)) + .toList()); + } + } + + @Mock + private KafkaAggregateRuntime delegate; + + @Mock + private AgentPlatform agentPlatform; + + @Mock + private AgentProcess agentProcess; + + @Mock + private AgentProcess distillerProcess; + + @Mock + private Blackboard blackboard; + + @Mock + private Blackboard distillerBlackboard; + + @Mock + private CommandBus commandBus; + + @Mock + private AgenticAggregate aggregate; + + @Mock + private Agent memoryDistillerAgent; + + private ObjectMapper objectMapper; + private KafkaAgenticAggregateRuntime runtime; + + @BeforeEach + void setUp() { + objectMapper = JsonMapper.builder().build(); + runtime = new KafkaAgenticAggregateRuntime( + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 100); + } + + // ------------------------------------------------------------------------- + // Memory distillation on COMPLETED process + // ------------------------------------------------------------------------- + + @Test + @SuppressWarnings("unchecked") + void resumeShouldDistillMemoriesOnCompletedProcess() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + // Process finishes with COMPLETED status + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + // Set up MemoryDistillerAgent resolution + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Set up distiller process + var distillationResult = new MemoryDistillationResult( + List.of(new MemoryDistillationResult.StoredMemory( + "testing", "Use JUnit 5", "src/test", "best practice")), + List.of()); + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(distillationResult); + + // Capture the events stream + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // Should have: tick events (empty) + AgentTaskFinishedEvent + MemoryStoredEvent + assertThat(events).hasSize(2); + assertThat(events.get(0)).isInstanceOf(AgentTaskFinishedEvent.class); + assertThat(events.get(1)).isInstanceOf(MemoryStoredEvent.class); + + MemoryStoredEvent storedEvent = (MemoryStoredEvent) events.get(1); + assertThat(storedEvent.agenticAggregateId()).isEqualTo("agg-1"); + assertThat(storedEvent.subject()).isEqualTo("testing"); + assertThat(storedEvent.fact()).isEqualTo("Use JUnit 5"); + + // Verify distiller process was run to completion + verify(distillerProcess).run(); + } + + @Test + @SuppressWarnings("unchecked") + void resumeShouldNotDistillMemoriesOnFailedProcess() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + // Process finishes with FAILED status + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.FAILED); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // Should only have AgentTaskFinishedEvent, no memory distillation + assertThat(events).hasSize(1); + assertThat(events.getFirst()).isInstanceOf(AgentTaskFinishedEvent.class); + + // Should NOT create a distiller process + verify(agentPlatform, never()).createAgentProcess( + argThat(a -> MemoryDistillerAgent.AGENT_NAME.equals(a.getName())), + any(), anyMap()); + } + + @Test + @SuppressWarnings("unchecked") + void resumeShouldSkipDistillationWhenDistillerAgentNotDeployed() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + + // No MemoryDistillerAgent deployed + when(agentPlatform.agents()).thenReturn(List.of()); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // Should only have AgentTaskFinishedEvent + assertThat(events).hasSize(1); + assertThat(events.getFirst()).isInstanceOf(AgentTaskFinishedEvent.class); + } + + @Test + @SuppressWarnings("unchecked") + void resumeShouldHandleDistillationFailureGracefully() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Distiller throws an exception + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenThrow(new RuntimeException("LLM unavailable")); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // Should still emit AgentTaskFinishedEvent despite distillation failure + assertThat(events).hasSize(1); + assertThat(events.getFirst()).isInstanceOf(AgentTaskFinishedEvent.class); + } + + // ------------------------------------------------------------------------- + // Memory limit enforcement + // ------------------------------------------------------------------------- + + @Test + @SuppressWarnings("unchecked") + void distillationShouldEnforceMaxMemoriesLimit() throws IOException { + // Set up runtime with maxMemories = 5 + runtime = new KafkaAgenticAggregateRuntime( + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5); + + // State already has 3 memories, so maxNewMemories = 2 + var existingMemories = List.of( + new AgenticAggregateMemory("mem-1", "s1", "f1", "c1", "r1", Instant.now()), + new AgenticAggregateMemory("mem-2", "s2", "f2", "c2", "r2", Instant.now()), + new AgenticAggregateMemory("mem-3", "s3", "f3", "c3", "r3", Instant.now()) + ); + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), existingMemories); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Agent tries to store 5 memories but only 2 should be allowed (maxMemories=5, existing=3) + var distillationResult = new MemoryDistillationResult( + List.of( + new MemoryDistillationResult.StoredMemory("s1", "fact1", "c1", "r1"), + new MemoryDistillationResult.StoredMemory("s2", "fact2", "c2", "r2"), + new MemoryDistillationResult.StoredMemory("s3", "fact3", "c3", "r3"), + new MemoryDistillationResult.StoredMemory("s4", "fact4", "c4", "r4"), + new MemoryDistillationResult.StoredMemory("s5", "fact5", "c5", "r5") + ), + List.of()); + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(distillationResult); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // 1 finished + 2 stored (truncated from 5) + assertThat(events).hasSize(3); + assertThat(events.get(0)).isInstanceOf(AgentTaskFinishedEvent.class); + long storedCount = events.stream().filter(e -> e instanceof MemoryStoredEvent).count(); + assertThat(storedCount).isEqualTo(2); + } + + @Test + @SuppressWarnings("unchecked") + void distillationShouldAllowMoreStoredWhenMemoriesAreRevoked() throws IOException { + // maxMemories = 5, existing = 4, so maxNew = 1 + // But we also revoke 2, so we can store up to 3 (1 + 2) + runtime = new KafkaAgenticAggregateRuntime( + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5); + + var existingMemories = List.of( + new AgenticAggregateMemory("mem-1", "s1", "f1", "c1", "r1", Instant.now()), + new AgenticAggregateMemory("mem-2", "s2", "f2", "c2", "r2", Instant.now()), + new AgenticAggregateMemory("mem-3", "s3", "f3", "c3", "r3", Instant.now()), + new AgenticAggregateMemory("mem-4", "s4", "f4", "c4", "r4", Instant.now()) + ); + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), existingMemories); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Revoke 2, store 3 + var distillationResult = new MemoryDistillationResult( + List.of( + new MemoryDistillationResult.StoredMemory("s1", "new-fact1", "c1", "r1"), + new MemoryDistillationResult.StoredMemory("s2", "new-fact2", "c2", "r2"), + new MemoryDistillationResult.StoredMemory("s3", "new-fact3", "c3", "r3") + ), + List.of( + new MemoryDistillationResult.RevokedMemory("mem-1", "outdated"), + new MemoryDistillationResult.RevokedMemory("mem-2", "superseded") + )); + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(distillationResult); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // 1 finished + 2 revoked + 3 stored = 6 + assertThat(events).hasSize(6); + assertThat(events.get(0)).isInstanceOf(AgentTaskFinishedEvent.class); + + long revokedCount = events.stream().filter(e -> e instanceof MemoryRevokedEvent).count(); + long storedCount = events.stream().filter(e -> e instanceof MemoryStoredEvent).count(); + assertThat(revokedCount).isEqualTo(2); + assertThat(storedCount).isEqualTo(3); + } + + @Test + @SuppressWarnings("unchecked") + void distillationShouldSkipRevocationOfNonExistentMemories() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var existingMemories = List.of( + new AgenticAggregateMemory("mem-1", "s1", "f1", "c1", "r1", Instant.now()) + ); + var state = new MemoryTaskState("agg-1", List.of(task), existingMemories); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Try to revoke a memory that doesn't exist + var distillationResult = new MemoryDistillationResult( + List.of(), + List.of( + new MemoryDistillationResult.RevokedMemory("mem-1", "outdated"), + new MemoryDistillationResult.RevokedMemory("non-existent", "cleanup") + )); + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(distillationResult); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // 1 finished + 1 revoked (only mem-1, not non-existent) + long revokedCount = events.stream().filter(e -> e instanceof MemoryRevokedEvent).count(); + assertThat(revokedCount).isEqualTo(1); + + MemoryRevokedEvent revokedEvent = events.stream() + .filter(e -> e instanceof MemoryRevokedEvent) + .map(e -> (MemoryRevokedEvent) e) + .findFirst() + .orElseThrow(); + assertThat(revokedEvent.memoryId()).isEqualTo("mem-1"); + } + + @Test + @SuppressWarnings("unchecked") + void distillationShouldHandleNullResultGracefully() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Distiller produces no result + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(null); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // Should only have AgentTaskFinishedEvent + assertThat(events).hasSize(1); + assertThat(events.getFirst()).isInstanceOf(AgentTaskFinishedEvent.class); + } + + @Test + @SuppressWarnings("unchecked") + void distillationShouldPassCorrectBindingsToDistillerProcess() throws IOException { + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var existingMemory = new AgenticAggregateMemory( + "mem-1", "testing", "Use JUnit", "cite", "reason", Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of(existingMemory)); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of("some-object")); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(null); + + @SuppressWarnings("rawtypes") + ArgumentCaptor> bindingsCaptor = ArgumentCaptor.forClass(Map.class); + + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(agentPlatform).createAgentProcess( + eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), bindingsCaptor.capture()); + + Map bindings = bindingsCaptor.getValue(); + assertThat(bindings).containsKey("history"); + assertThat(bindings).containsKey("blackboardObjects"); + assertThat(bindings).containsKey("existingMemories"); + assertThat(bindings).containsKey("maxNewMemories"); + + assertThat(bindings.get("existingMemories")).isEqualTo(List.of(existingMemory)); + // maxMemories=100, existing=1, so maxNewMemories=99 + assertThat(bindings.get("maxNewMemories")).isEqualTo(99); + } +} diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java index cb4d1b78..39a74462 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java @@ -113,7 +113,7 @@ public String getAggregateId() { @BeforeEach void setUp() { objectMapper = JsonMapper.builder().build(); - runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TaskState.class, agentPlatform, aggregate); + runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TaskState.class, agentPlatform, aggregate, 100); } @Test @@ -126,7 +126,7 @@ void resumeShouldDoNothingWhenStateIsNull() throws IOException { @Test void resumeShouldDoNothingWhenStateDoesNotImplementTaskAwareState() throws IOException { - var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate); + var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100); var state = new SimpleState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); var stateRecord = new AggregateStateRecord(null, "SimpleState", 1, payload, @@ -235,7 +235,7 @@ void hasActiveAgentTasksShouldReturnFalseWhenStateIsNull() throws IOException { @Test void hasActiveAgentTasksShouldReturnFalseWhenNotTaskAware() throws IOException { - var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate); + var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100); var state = new SimpleState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); var stateRecord = new AggregateStateRecord(null, "SimpleState", 1, payload, From 29b7045fb4d2e4cb98f50d666660fd39f246ae76 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 11:34:26 +0000 Subject: [PATCH 02/10] Address code review: extract local variable, remove redundant null check Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/b0b7907c-8588-4b46-af3a-67b710ba24ad Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akces/agentic/embabel/MemoryDistillerAgent.java | 13 +++++-------- .../runtime/KafkaAgenticAggregateRuntime.java | 2 +- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 91dcf951..6a265dcc 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -78,14 +78,11 @@ public class MemoryDistillerAgent { @Action(description = "Distill relevant memories from a completed agent process") @AchievesGoal(description = "Distill and manage memories from completed agent processes") public MemoryDistillationResult distillMemories(ActionContext context) { - List history = context.getProcessContext().getAgentProcess() - .getBlackboard().objectsOfType(ActionInvocation.class); - List blackboardObjects = (List) context.getProcessContext() - .getAgentProcess().getBlackboard().get("blackboardObjects"); - List existingMemories = (List) context.getProcessContext() - .getAgentProcess().getBlackboard().get("existingMemories"); - int maxNewMemories = (int) context.getProcessContext() - .getAgentProcess().getBlackboard().get("maxNewMemories"); + var blackboard = context.getProcessContext().getAgentProcess().getBlackboard(); + List history = blackboard.objectsOfType(ActionInvocation.class); + List blackboardObjects = (List) blackboard.get("blackboardObjects"); + List existingMemories = (List) blackboard.get("existingMemories"); + int maxNewMemories = (int) blackboard.get("maxNewMemories"); String prompt = buildPrompt(history, blackboardObjects, existingMemories, maxNewMemories); diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index cc7324ac..50194c23 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -638,7 +638,7 @@ private List translateDistillationResult(MemoryDistillationResult r validRevocations.add(revoked); } else { logger.debug("Skipping revocation of non-existent memory '{}' for aggregate {}", - revoked != null ? revoked.memoryId() : "null", getName()); + revoked.memoryId(), getName()); } } } From 6aa01aa17691e6287c4d6fd582b515ce8d3dec35 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 11:58:31 +0000 Subject: [PATCH 03/10] Address PR review feedback: rename params, use events directly, rename agent names, multi-line prompt, warn log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename maxMemories to maxTotalMemories and add maxMemoriesAdded in @AgenticAggregateInfo - Use MemoryStoredEvent/MemoryRevokedEvent directly in MemoryDistillationResult (no intermediate DTOs) - Rename agent names: DefaultAgent → "Default", MemoryDistillerAgent → "MemoryDistiller" - Use multi-line String (text blocks) in MemoryDistillerAgent prompt building - Change "not deployed" log level from DEBUG to WARN - Pass both maxTotalMemories and maxMemoriesAdded to distiller bindings - De-duplicate revocations by memoryId - Add safe binding reads with instanceof checks - Update all tests for new signatures and types Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/40e6b583-f254-41d2-95c4-718e34edd19e Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../beans/AgenticAggregateRuntimeFactory.java | 2 +- .../akces/agentic/embabel/DefaultAgent.java | 2 +- .../embabel/MemoryDistillationResult.java | 46 ++------ .../agentic/embabel/MemoryDistillerAgent.java | 89 ++++++++++----- .../agentic/events/MemoryRevokedEvent.java | 2 +- .../runtime/KafkaAgenticAggregateRuntime.java | 80 ++++++------- .../embabel/MemoryDistillationResultTest.java | 37 ++---- .../AgenticAggregateAutoCreateTest.java | 6 +- .../KafkaAgenticAggregateRuntimeTest.java | 14 +-- .../runtime/MemoryDistillationTest.java | 107 ++++++++++++++---- .../runtime/ResumeNextAgentTaskTest.java | 6 +- .../annotations/AgenticAggregateInfo.java | 12 +- .../aggregate/AgenticAggregateInfoTest.java | 22 +++- .../EventCatalogProcessorTest.java | 2 +- 14 files changed, 251 insertions(+), 176 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java index 1361ab2e..46a73823 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java @@ -135,7 +135,7 @@ public AgenticAggregateRuntime getObject() { KafkaAggregateRuntime kafkaRuntime = createRuntime(agenticInfo, aggregate, agentPlatform); return new KafkaAgenticAggregateRuntime( kafkaRuntime, objectMapper, agenticInfo.stateClass(), agentPlatform, aggregate, - agenticInfo.maxMemories()); + agenticInfo.maxTotalMemories(), agenticInfo.maxMemoriesAdded()); } @Override diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java index 7d2c1618..f9d28569 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java @@ -35,5 +35,5 @@ public class DefaultAgent { * The well-known agent name used to locate this default agent on the * {@link com.embabel.agent.core.AgentPlatform} during fallback resolution. */ - public static final String AGENT_NAME = "DefaultAgent"; + public static final String AGENT_NAME = "Default"; } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java index a50b53a5..873a7280 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResult.java @@ -17,6 +17,9 @@ package org.elasticsoftware.akces.agentic.embabel; +import org.elasticsoftware.akces.agentic.events.MemoryRevokedEvent; +import org.elasticsoftware.akces.agentic.events.MemoryStoredEvent; + import java.util.List; /** @@ -25,46 +28,15 @@ * *

Contains two lists: *

    - *
  • {@code stored} — new memory entries to be persisted as - * {@link org.elasticsoftware.akces.agentic.events.MemoryStoredEvent}s
    • - *
  • {@code revoked} — existing memory entries to be removed as - * {@link org.elasticsoftware.akces.agentic.events.MemoryRevokedEvent}s
    • + *
  • {@code stored} — new memory entries as {@link MemoryStoredEvent}s
    • + *
  • {@code revoked} — memory entries to remove as {@link MemoryRevokedEvent}s
    • *
* - * @param stored the list of memories to store; each entry contains subject, fact, - * citations, and reason fields - * @param revoked the list of memories to revoke; each entry contains memoryId and reason + * @param stored the list of memories to store + * @param revoked the list of memories to revoke */ public record MemoryDistillationResult( - List stored, - List revoked + List stored, + List revoked ) { - - /** - * A memory entry to be stored. - * - * @param subject a short (1–2 word) topic label - * @param fact the fact to remember (max 200 characters) - * @param citations the source citation for the fact - * @param reason why this memory should be stored - */ - public record StoredMemory( - String subject, - String fact, - String citations, - String reason - ) { - } - - /** - * A memory entry to be revoked. - * - * @param memoryId the UUID of the memory to revoke - * @param reason why this memory should be revoked - */ - public record RevokedMemory( - String memoryId, - String reason - ) { - } } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 6a265dcc..60efc065 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -22,7 +22,6 @@ import com.embabel.agent.api.annotation.Agent; import com.embabel.agent.api.common.ActionContext; import com.embabel.agent.api.common.PlannerType; -import com.embabel.agent.core.ActionInvocation; import java.util.List; @@ -41,9 +40,9 @@ *
  • A list of existing memories to revoke
    • * * - *

    The net number of new memories (stored minus revoked) is constrained by the - * {@code maxNewMemories} binding on the blackboard, which is derived from the - * aggregate's {@code maxMemories} configuration minus the current memory count. + *

    The net number of new memories (stored minus revoked) is constrained by both + * the {@code maxTotalMemories} capacity and the per-distillation budget + * {@code maxMemoriesAdded}. */ @Agent(name = MemoryDistillerAgent.AGENT_NAME, description = "Distills relevant memories from a completed agent process", @@ -54,7 +53,7 @@ public class MemoryDistillerAgent { * The well-known agent name used to locate this agent on the * {@link com.embabel.agent.core.AgentPlatform}. */ - public static final String AGENT_NAME = "MemoryDistillerAgent"; + public static final String AGENT_NAME = "MemoryDistiller"; /** * Distills memories from a completed agent process by analyzing the process @@ -62,14 +61,13 @@ public class MemoryDistillerAgent { * *

    The method reads the following bindings from the blackboard: *

      - *
    • {@code "history"} — the {@link List} of {@link ActionInvocation}s from the - * completed process
      • + *
    • {@code "history"} — the execution history from the completed process
      • *
    • {@code "blackboardObjects"} — all objects from the completed process's * blackboard
      • *
    • {@code "existingMemories"} — the current list of memories from the aggregate * state
      • - *
    • {@code "maxNewMemories"} — the maximum number of net new memories allowed - * (maxMemories - currentMemoryCount)
      • + *
    • {@code "maxTotalMemories"} — the total memory capacity of the aggregate
      • + *
    • {@code "maxMemoriesAdded"} — the per-distillation budget for net new memories
      • *
    * * @param context the action context providing access to AI capabilities and blackboard @@ -79,12 +77,33 @@ public class MemoryDistillerAgent { @AchievesGoal(description = "Distill and manage memories from completed agent processes") public MemoryDistillationResult distillMemories(ActionContext context) { var blackboard = context.getProcessContext().getAgentProcess().getBlackboard(); - List history = blackboard.objectsOfType(ActionInvocation.class); - List blackboardObjects = (List) blackboard.get("blackboardObjects"); - List existingMemories = (List) blackboard.get("existingMemories"); - int maxNewMemories = (int) blackboard.get("maxNewMemories"); - String prompt = buildPrompt(history, blackboardObjects, existingMemories, maxNewMemories); + Object historyBinding = blackboard.get("history"); + List history = historyBinding instanceof List historyList + ? historyList + : List.of(); + Object blackboardObjectsBinding = blackboard.get("blackboardObjects"); + List blackboardObjects = blackboardObjectsBinding instanceof List objectList + ? objectList + : List.of(); + Object existingMemoriesBinding = blackboard.get("existingMemories"); + List existingMemories = existingMemoriesBinding instanceof List memoryList + ? memoryList + : List.of(); + Object maxTotalMemoriesBinding = blackboard.get("maxTotalMemories"); + int maxTotalMemories = maxTotalMemoriesBinding instanceof Number number + ? number.intValue() + : 0; + Object maxMemoriesAddedBinding = blackboard.get("maxMemoriesAdded"); + int maxMemoriesAdded = maxMemoriesAddedBinding instanceof Number number + ? number.intValue() + : 0; + + int currentCount = existingMemories.size(); + int capacityLeft = Math.max(0, maxTotalMemories - currentCount); + int effectiveLimit = Math.min(capacityLeft, maxMemoriesAdded); + + String prompt = buildPrompt(history, blackboardObjects, existingMemories, effectiveLimit); return context.ai() .withDefaultLlm() @@ -95,21 +114,31 @@ private String buildPrompt(List history, List blackboardObjects, List existingMemories, int maxNewMemories) { - StringBuilder sb = new StringBuilder(); - sb.append("You are a memory distillation agent. Analyze the following completed agent process "); - sb.append("and determine which facts should be stored as new memories and which existing "); - sb.append("memories should be revoked (because they are outdated, incorrect, or superseded).\n\n"); + var sb = new StringBuilder(); + + sb.append(""" + You are a memory distillation agent. Analyze the following completed agent process \ + and determine which facts should be stored as new memories and which existing \ + memories should be revoked (because they are outdated, incorrect, or superseded). + + CONSTRAINTS: + """); - sb.append("CONSTRAINTS:\n"); sb.append("- The net number of new memories (stored count minus revoked count) must be <= ") .append(maxNewMemories).append("\n"); - sb.append("- Each stored memory must have: subject (1-2 words), fact (max 200 chars), "); - sb.append("citations (source reference), and reason (why it should be stored)\n"); - sb.append("- Each revoked memory must have: memoryId (from existing memories) and reason\n"); - sb.append("- Only store memories that are actionable, likely to remain relevant, and "); - sb.append("cannot always be inferred from limited context\n"); - sb.append("- Only revoke memories that are clearly outdated, incorrect, or superseded "); - sb.append("by new information\n\n"); + + sb.append(""" + - Each stored memory must have: agenticAggregateId, memoryId (UUID), \ + subject (1-2 words), fact (max 200 chars), citations (source reference), \ + reason (why it should be stored), and storedAt (ISO-8601 instant) + - Each revoked memory must have: agenticAggregateId, memoryId (from existing \ + memories), reason, and revokedAt (ISO-8601 instant) + - Only store memories that are actionable, likely to remain relevant, and \ + cannot always be inferred from limited context + - Only revoke memories that are clearly outdated, incorrect, or superseded \ + by new information + + """); sb.append("PROCESS HISTORY (actions executed):\n"); if (history != null && !history.isEmpty()) { @@ -139,9 +168,11 @@ private String buildPrompt(List history, sb.append("(no existing memories)\n"); } - sb.append("\nReturn a JSON object with 'stored' (list of new memories to store) "); - sb.append("and 'revoked' (list of existing memories to revoke). "); - sb.append("If no memories should be stored or revoked, return empty lists."); + sb.append(""" + + Return a JSON object with 'stored' (list of MemoryStoredEvent instances) \ + and 'revoked' (list of MemoryRevokedEvent instances). \ + If no memories should be stored or revoked, return empty lists."""); return sb.toString(); } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/events/MemoryRevokedEvent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/events/MemoryRevokedEvent.java index c3599c94..8330c2e5 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/events/MemoryRevokedEvent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/events/MemoryRevokedEvent.java @@ -31,7 +31,7 @@ *

    This event is produced internally by the Embabel layer (via the agent's memory * management tools) when a memory is explicitly revoked, or as part of the * sliding-window eviction mechanism when - * {@link org.elasticsoftware.akces.annotations.AgenticAggregateInfo#maxMemories()} is exceeded. + * {@link org.elasticsoftware.akces.annotations.AgenticAggregateInfo#maxTotalMemories()} is exceeded. * * @param agenticAggregateId the unique identifier of the AgenticAggregate instance * @param memoryId UUID of the memory entry that was revoked diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index 50194c23..d2fb923a 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -73,7 +73,8 @@ public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime { private final Class stateClass; private final AgentPlatform agentPlatform; private final AgenticAggregate aggregate; - private final int maxMemories; + private final int maxTotalMemories; + private final int maxMemoriesAdded; /** Round-robin counter for selecting the next agent task to resume. */ private final AtomicInteger nextTaskIndex = new AtomicInteger(0); @@ -81,28 +82,31 @@ public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime { /** * Creates a new {@code KafkaAgenticAggregateRuntime}. * - * @param delegate the underlying aggregate runtime to delegate to - * @param objectMapper the Jackson {@link ObjectMapper} used for JSON serialization - * @param stateClass the aggregate state class - * @param agentPlatform the Embabel {@link AgentPlatform} used for AI-assisted processing; - * must not be {@code null} - * @param aggregate the agentic aggregate instance whose - * {@link AgenticAggregate#getCreateDomainEvent()} method provides the - * auto-create event - * @param maxMemories the maximum number of memories allowed for this aggregate + * @param delegate the underlying aggregate runtime to delegate to + * @param objectMapper the Jackson {@link ObjectMapper} used for JSON serialization + * @param stateClass the aggregate state class + * @param agentPlatform the Embabel {@link AgentPlatform} used for AI-assisted processing; + * must not be {@code null} + * @param aggregate the agentic aggregate instance whose + * {@link AgenticAggregate#getCreateDomainEvent()} method provides the + * auto-create event + * @param maxTotalMemories the total memory capacity for this aggregate + * @param maxMemoriesAdded the per-distillation budget for net new memories */ public KafkaAgenticAggregateRuntime(KafkaAggregateRuntime delegate, ObjectMapper objectMapper, Class stateClass, AgentPlatform agentPlatform, AgenticAggregate aggregate, - int maxMemories) { + int maxTotalMemories, + int maxMemoriesAdded) { this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); this.stateClass = Objects.requireNonNull(stateClass, "stateClass must not be null"); this.agentPlatform = Objects.requireNonNull(agentPlatform, "agentPlatform must not be null"); this.aggregate = Objects.requireNonNull(aggregate, "aggregate must not be null"); - this.maxMemories = maxMemories; + this.maxTotalMemories = maxTotalMemories; + this.maxMemoriesAdded = maxMemoriesAdded; } // ------------------------------------------------------------------------- @@ -555,20 +559,20 @@ public void resumeNextAgentTask(Consumer protocolRecordConsumer, private List distillMemories(AgentProcess completedProcess, AggregateState state) { Agent memoryDistillerAgent = resolveMemoryDistillerAgent(); if (memoryDistillerAgent == null) { - logger.debug("MemoryDistillerAgent not deployed on the platform; skipping memory distillation"); + logger.warn("MemoryDistillerAgent not deployed on the platform; skipping memory distillation"); return List.of(); } List currentMemories = state instanceof MemoryAwareState mas ? mas.getMemories() : List.of(); - int maxNewMemories = Math.max(0, maxMemories - currentMemories.size()); Map bindings = new LinkedHashMap<>(); bindings.put("history", completedProcess.getHistory()); bindings.put("blackboardObjects", completedProcess.getBlackboard().getObjects()); bindings.put("existingMemories", currentMemories); - bindings.put("maxNewMemories", maxNewMemories); + bindings.put("maxTotalMemories", maxTotalMemories); + bindings.put("maxMemoriesAdded", maxMemoriesAdded); try { AgentProcess distillerProcess = agentPlatform.createAgentProcess( @@ -583,7 +587,9 @@ private List distillMemories(AgentProcess completedProcess, Aggrega return List.of(); } - return translateDistillationResult(result, state.getAggregateId(), currentMemories, maxNewMemories); + int capacityLeft = Math.max(0, maxTotalMemories - currentMemories.size()); + int effectiveLimit = Math.min(capacityLeft, maxMemoriesAdded); + return translateDistillationResult(result, state.getAggregateId(), currentMemories, effectiveLimit); } catch (Exception e) { logger.warn("Memory distillation failed for aggregate {}; proceeding without memory updates", getName(), e); @@ -623,53 +629,51 @@ private List translateDistillationResult(MemoryDistillationResult r List currentMemories, int maxNewMemories) { List events = new ArrayList<>(); - Instant now = Instant.now(); - // Collect valid revocations first (only for memories that actually exist) + // Collect valid revocations first (only for memories that actually exist), + // de-duplicated by memoryId so capacity calculations match effective revocations. Set currentMemoryIds = new HashSet<>(); for (AgenticAggregateMemory mem : currentMemories) { currentMemoryIds.add(mem.memoryId()); } - List validRevocations = new ArrayList<>(); + List validRevocations = new ArrayList<>(); + Set revokedMemoryIds = new HashSet<>(); if (result.revoked() != null) { - for (MemoryDistillationResult.RevokedMemory revoked : result.revoked()) { - if (revoked.memoryId() != null && currentMemoryIds.contains(revoked.memoryId())) { - validRevocations.add(revoked); - } else { + for (MemoryRevokedEvent revoked : result.revoked()) { + String memoryId = revoked.memoryId(); + if (memoryId == null || !currentMemoryIds.contains(memoryId)) { logger.debug("Skipping revocation of non-existent memory '{}' for aggregate {}", - revoked.memoryId(), getName()); + memoryId, getName()); + continue; } + if (!revokedMemoryIds.add(memoryId)) { + logger.debug("Skipping duplicate revocation of memory '{}' for aggregate {}", + memoryId, getName()); + continue; + } + validRevocations.add(revoked); } } // Calculate how many stored memories we can accept - int revokedCount = validRevocations.size(); + int revokedCount = revokedMemoryIds.size(); int maxStoredCount = maxNewMemories + revokedCount; // Add revocation events - for (MemoryDistillationResult.RevokedMemory revoked : validRevocations) { - events.add(new MemoryRevokedEvent(aggregateId, revoked.memoryId(), revoked.reason(), now)); - } + events.addAll(validRevocations); // Add stored events (respecting the limit) if (result.stored() != null) { int storedCount = 0; - for (MemoryDistillationResult.StoredMemory stored : result.stored()) { + for (MemoryStoredEvent stored : result.stored()) { if (storedCount >= maxStoredCount) { logger.debug("Memory distillation limit reached ({} stored, {} revoked, max {}); " + "truncating remaining stored memories for aggregate {}", - storedCount, revokedCount, maxNewMemories, getName()); + storedCount, revokedCount, maxStoredCount, getName()); break; } - events.add(new MemoryStoredEvent( - aggregateId, - UUID.randomUUID().toString(), - stored.subject(), - stored.fact(), - stored.citations(), - stored.reason(), - now)); + events.add(stored); storedCount++; } } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java index aa4d1eea..4da92c90 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationResultTest.java @@ -17,8 +17,11 @@ package org.elasticsoftware.akces.agentic.embabel; +import org.elasticsoftware.akces.agentic.events.MemoryRevokedEvent; +import org.elasticsoftware.akces.agentic.events.MemoryStoredEvent; import org.junit.jupiter.api.Test; +import java.time.Instant; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -28,23 +31,24 @@ */ class MemoryDistillationResultTest { + private static final Instant NOW = Instant.now(); + @Test - void shouldCreateWithStoredAndRevokedMemories() { + void shouldCreateWithStoredAndRevokedEvents() { var stored = List.of( - new MemoryDistillationResult.StoredMemory("testing", "Use JUnit 5", "src/test", "best practice"), - new MemoryDistillationResult.StoredMemory("logging", "Use SLF4J", "src/main", "consistency") + new MemoryStoredEvent("agg-1", "mem-1", "testing", "Use JUnit 5", + "src/test", "best practice", NOW), + new MemoryStoredEvent("agg-1", "mem-2", "logging", "Use SLF4J", + "src/main", "consistency", NOW) ); var revoked = List.of( - new MemoryDistillationResult.RevokedMemory("mem-old-1", "superseded by new info") + new MemoryRevokedEvent("agg-1", "mem-old-1", "superseded by new info", NOW) ); var result = new MemoryDistillationResult(stored, revoked); assertThat(result.stored()).hasSize(2); assertThat(result.revoked()).hasSize(1); - assertThat(result.stored().getFirst().subject()).isEqualTo("testing"); - assertThat(result.stored().getFirst().fact()).isEqualTo("Use JUnit 5"); - assertThat(result.revoked().getFirst().memoryId()).isEqualTo("mem-old-1"); } @Test @@ -54,23 +58,4 @@ void shouldCreateWithEmptyLists() { assertThat(result.stored()).isEmpty(); assertThat(result.revoked()).isEmpty(); } - - @Test - void storedMemoryShouldExposeAllFields() { - var stored = new MemoryDistillationResult.StoredMemory( - "conventions", "Use records for DTOs", "src/main/Model.java:10", "immutability"); - - assertThat(stored.subject()).isEqualTo("conventions"); - assertThat(stored.fact()).isEqualTo("Use records for DTOs"); - assertThat(stored.citations()).isEqualTo("src/main/Model.java:10"); - assertThat(stored.reason()).isEqualTo("immutability"); - } - - @Test - void revokedMemoryShouldExposeAllFields() { - var revoked = new MemoryDistillationResult.RevokedMemory("mem-123", "outdated"); - - assertThat(revoked.memoryId()).isEqualTo("mem-123"); - assertThat(revoked.reason()).isEqualTo("outdated"); - } } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java index 8b14b8e4..db4172c2 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java @@ -240,7 +240,7 @@ void initializeStateShouldCallGetCreateDomainEventAndDelegateToRuntime() throws ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100); + delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100, 10); when(delegate.getName()).thenReturn("TestBot"); @@ -275,7 +275,7 @@ public Class getStateClass() { ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, nullReturningAggregate, 100); + delegate, objectMapper, TestBotState.class, agentPlatform, nullReturningAggregate, 100, 10); assertThatNullPointerException() .isThrownBy(() -> runtime.initializeState(pr -> {}, (der, ip) -> {})) @@ -289,7 +289,7 @@ void initializeStateShouldPassConsumersThroughToDelegate() throws IOException { ObjectMapper objectMapper = JsonMapper.builder().build(); KafkaAgenticAggregateRuntime runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100); + delegate, objectMapper, TestBotState.class, agentPlatform, testBot, 100, 10); when(delegate.getName()).thenReturn("TestBot"); diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java index 8f94e5a4..af435135 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java @@ -108,7 +108,7 @@ public String getAggregateId() { @BeforeEach void setUp() { objectMapper = JsonMapper.builder().build(); - runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100); + runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100, 10); } // ------------------------------------------------------------------------- @@ -145,7 +145,7 @@ void getMemoriesShouldReturnMemoriesFromMemoryAwareState() throws IOException { @Test void getMemoriesShouldReturnEmptyListForNonMemoryAwareState() throws IOException { - var plainRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, PlainState.class, agentPlatform, aggregate, 100); + var plainRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, PlainState.class, agentPlatform, aggregate, 100, 10); var state = new PlainState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); @@ -235,7 +235,7 @@ void getAgentPlatformShouldReturnInjectedPlatform() { void constructorShouldRejectNullDelegate() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - null, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100)) + null, objectMapper, TestMemoryState.class, agentPlatform, aggregate, 100, 10)) .withMessageContaining("delegate"); } @@ -243,7 +243,7 @@ void constructorShouldRejectNullDelegate() { void constructorShouldRejectNullObjectMapper() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, null, TestMemoryState.class, agentPlatform, aggregate, 100)) + delegate, null, TestMemoryState.class, agentPlatform, aggregate, 100, 10)) .withMessageContaining("objectMapper"); } @@ -251,7 +251,7 @@ void constructorShouldRejectNullObjectMapper() { void constructorShouldRejectNullStateClass() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, null, agentPlatform, aggregate, 100)) + delegate, objectMapper, null, agentPlatform, aggregate, 100, 10)) .withMessageContaining("stateClass"); } @@ -259,7 +259,7 @@ void constructorShouldRejectNullStateClass() { void constructorShouldRejectNullAgentPlatform() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestMemoryState.class, null, aggregate, 100)) + delegate, objectMapper, TestMemoryState.class, null, aggregate, 100, 10)) .withMessageContaining("agentPlatform"); } @@ -267,7 +267,7 @@ void constructorShouldRejectNullAgentPlatform() { void constructorShouldRejectNullAggregate() { assertThatNullPointerException() .isThrownBy(() -> new KafkaAgenticAggregateRuntime( - delegate, objectMapper, TestMemoryState.class, agentPlatform, null, 100)) + delegate, objectMapper, TestMemoryState.class, agentPlatform, null, 100, 10)) .withMessageContaining("aggregate"); } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java index 2ad98564..8fe54ddd 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java @@ -112,6 +112,8 @@ public MemoryAwareState withoutMemory(String memoryId) { } } + private static final Instant NOW = Instant.now(); + @Mock private KafkaAggregateRuntime delegate; @@ -146,7 +148,7 @@ public MemoryAwareState withoutMemory(String memoryId) { void setUp() { objectMapper = JsonMapper.builder().build(); runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 100); + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 100, 10); } // ------------------------------------------------------------------------- @@ -180,8 +182,8 @@ void resumeShouldDistillMemoriesOnCompletedProcess() throws IOException { // Set up distiller process var distillationResult = new MemoryDistillationResult( - List.of(new MemoryDistillationResult.StoredMemory( - "testing", "Use JUnit 5", "src/test", "best practice")), + List.of(new MemoryStoredEvent("agg-1", "mem-new-1", "testing", "Use JUnit 5", + "src/test", "best practice", NOW)), List.of()); when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) .thenReturn(distillerProcess); @@ -323,11 +325,11 @@ void resumeShouldHandleDistillationFailureGracefully() throws IOException { @Test @SuppressWarnings("unchecked") void distillationShouldEnforceMaxMemoriesLimit() throws IOException { - // Set up runtime with maxMemories = 5 + // maxTotalMemories = 5, maxMemoriesAdded = 10 — capacity is the binding constraint runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5); + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5, 10); - // State already has 3 memories, so maxNewMemories = 2 + // State already has 3 memories, so capacityLeft = 2, effectiveLimit = min(2, 10) = 2 var existingMemories = List.of( new AgenticAggregateMemory("mem-1", "s1", "f1", "c1", "r1", Instant.now()), new AgenticAggregateMemory("mem-2", "s2", "f2", "c2", "r2", Instant.now()), @@ -353,14 +355,14 @@ void distillationShouldEnforceMaxMemoriesLimit() throws IOException { when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); - // Agent tries to store 5 memories but only 2 should be allowed (maxMemories=5, existing=3) + // Agent tries to store 5 memories but only 2 should be allowed var distillationResult = new MemoryDistillationResult( List.of( - new MemoryDistillationResult.StoredMemory("s1", "fact1", "c1", "r1"), - new MemoryDistillationResult.StoredMemory("s2", "fact2", "c2", "r2"), - new MemoryDistillationResult.StoredMemory("s3", "fact3", "c3", "r3"), - new MemoryDistillationResult.StoredMemory("s4", "fact4", "c4", "r4"), - new MemoryDistillationResult.StoredMemory("s5", "fact5", "c5", "r5") + new MemoryStoredEvent("agg-1", "n1", "s1", "fact1", "c1", "r1", NOW), + new MemoryStoredEvent("agg-1", "n2", "s2", "fact2", "c2", "r2", NOW), + new MemoryStoredEvent("agg-1", "n3", "s3", "fact3", "c3", "r3", NOW), + new MemoryStoredEvent("agg-1", "n4", "s4", "fact4", "c4", "r4", NOW), + new MemoryStoredEvent("agg-1", "n5", "s5", "fact5", "c5", "r5", NOW) ), List.of()); when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) @@ -381,13 +383,68 @@ void distillationShouldEnforceMaxMemoriesLimit() throws IOException { assertThat(storedCount).isEqualTo(2); } + @Test + @SuppressWarnings("unchecked") + void distillationShouldEnforceMaxMemoriesAddedLimit() throws IOException { + // maxTotalMemories = 100, maxMemoriesAdded = 2 — per-distillation budget is the constraint + runtime = new KafkaAgenticAggregateRuntime( + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 100, 2); + + var party = new HumanRequestingParty("user-1", "analyst"); + var task = new AssignedTask("proc-1", "Task", party, Map.of(), Instant.now()); + var state = new MemoryTaskState("agg-1", List.of(task), List.of()); + byte[] payload = objectMapper.writeValueAsBytes(state); + var stateRecord = new AggregateStateRecord(null, "MemoryTaskState", 1, payload, + PayloadEncoding.JSON, "agg-1", null, 1L); + when(delegate.materializeState(stateRecord)).thenReturn(state); + + when(agentPlatform.getAgentProcess("proc-1")).thenReturn(agentProcess); + when(agentProcess.getBlackboard()).thenReturn(blackboard); + when(blackboard.getObjects()).thenReturn(List.of()); + when(delegate.getAllDomainEventTypes()).thenReturn(List.of()); + + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(AgentProcessStatusCode.COMPLETED); + when(agentProcess.getHistory()).thenReturn(List.of()); + + when(memoryDistillerAgent.getName()).thenReturn(MemoryDistillerAgent.AGENT_NAME); + when(agentPlatform.agents()).thenReturn(List.of(memoryDistillerAgent)); + + // Agent tries to store 5 memories but only 2 should be allowed (maxMemoriesAdded=2) + var distillationResult = new MemoryDistillationResult( + List.of( + new MemoryStoredEvent("agg-1", "n1", "s1", "fact1", "c1", "r1", NOW), + new MemoryStoredEvent("agg-1", "n2", "s2", "fact2", "c2", "r2", NOW), + new MemoryStoredEvent("agg-1", "n3", "s3", "fact3", "c3", "r3", NOW), + new MemoryStoredEvent("agg-1", "n4", "s4", "fact4", "c4", "r4", NOW), + new MemoryStoredEvent("agg-1", "n5", "s5", "fact5", "c5", "r5", NOW) + ), + List.of()); + when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) + .thenReturn(distillerProcess); + when(distillerProcess.getBlackboard()).thenReturn(distillerBlackboard); + when(distillerBlackboard.last(MemoryDistillationResult.class)).thenReturn(distillationResult); + + ArgumentCaptor> eventsCaptor = ArgumentCaptor.forClass(Stream.class); + runtime.resumeNextAgentTask(pr -> {}, () -> stateRecord, commandBus); + + verify(delegate).processDomainEvents(eventsCaptor.capture(), eq("proc-1"), any(), any()); + List events = eventsCaptor.getValue().toList(); + + // 1 finished + 2 stored (truncated from 5 by maxMemoriesAdded) + assertThat(events).hasSize(3); + assertThat(events.get(0)).isInstanceOf(AgentTaskFinishedEvent.class); + long storedCount = events.stream().filter(e -> e instanceof MemoryStoredEvent).count(); + assertThat(storedCount).isEqualTo(2); + } + @Test @SuppressWarnings("unchecked") void distillationShouldAllowMoreStoredWhenMemoriesAreRevoked() throws IOException { - // maxMemories = 5, existing = 4, so maxNew = 1 + // maxTotalMemories = 5, maxMemoriesAdded = 10; existing = 4, so capacityLeft = 1 // But we also revoke 2, so we can store up to 3 (1 + 2) runtime = new KafkaAgenticAggregateRuntime( - delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5); + delegate, objectMapper, MemoryTaskState.class, agentPlatform, aggregate, 5, 10); var existingMemories = List.of( new AgenticAggregateMemory("mem-1", "s1", "f1", "c1", "r1", Instant.now()), @@ -418,13 +475,13 @@ void distillationShouldAllowMoreStoredWhenMemoriesAreRevoked() throws IOExceptio // Revoke 2, store 3 var distillationResult = new MemoryDistillationResult( List.of( - new MemoryDistillationResult.StoredMemory("s1", "new-fact1", "c1", "r1"), - new MemoryDistillationResult.StoredMemory("s2", "new-fact2", "c2", "r2"), - new MemoryDistillationResult.StoredMemory("s3", "new-fact3", "c3", "r3") + new MemoryStoredEvent("agg-1", "n1", "s1", "new-fact1", "c1", "r1", NOW), + new MemoryStoredEvent("agg-1", "n2", "s2", "new-fact2", "c2", "r2", NOW), + new MemoryStoredEvent("agg-1", "n3", "s3", "new-fact3", "c3", "r3", NOW) ), List.of( - new MemoryDistillationResult.RevokedMemory("mem-1", "outdated"), - new MemoryDistillationResult.RevokedMemory("mem-2", "superseded") + new MemoryRevokedEvent("agg-1", "mem-1", "outdated", NOW), + new MemoryRevokedEvent("agg-1", "mem-2", "superseded", NOW) )); when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) .thenReturn(distillerProcess); @@ -477,8 +534,8 @@ void distillationShouldSkipRevocationOfNonExistentMemories() throws IOException var distillationResult = new MemoryDistillationResult( List.of(), List.of( - new MemoryDistillationResult.RevokedMemory("mem-1", "outdated"), - new MemoryDistillationResult.RevokedMemory("non-existent", "cleanup") + new MemoryRevokedEvent("agg-1", "mem-1", "outdated", NOW), + new MemoryRevokedEvent("agg-1", "non-existent", "cleanup", NOW) )); when(agentPlatform.createAgentProcess(eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), anyMap())) .thenReturn(distillerProcess); @@ -585,10 +642,12 @@ void distillationShouldPassCorrectBindingsToDistillerProcess() throws IOExceptio assertThat(bindings).containsKey("history"); assertThat(bindings).containsKey("blackboardObjects"); assertThat(bindings).containsKey("existingMemories"); - assertThat(bindings).containsKey("maxNewMemories"); + assertThat(bindings).containsKey("maxTotalMemories"); + assertThat(bindings).containsKey("maxMemoriesAdded"); assertThat(bindings.get("existingMemories")).isEqualTo(List.of(existingMemory)); - // maxMemories=100, existing=1, so maxNewMemories=99 - assertThat(bindings.get("maxNewMemories")).isEqualTo(99); + // maxTotalMemories=100, maxMemoriesAdded=10 + assertThat(bindings.get("maxTotalMemories")).isEqualTo(100); + assertThat(bindings.get("maxMemoriesAdded")).isEqualTo(10); } } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java index 39a74462..3f9855d4 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/ResumeNextAgentTaskTest.java @@ -113,7 +113,7 @@ public String getAggregateId() { @BeforeEach void setUp() { objectMapper = JsonMapper.builder().build(); - runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TaskState.class, agentPlatform, aggregate, 100); + runtime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, TaskState.class, agentPlatform, aggregate, 100, 10); } @Test @@ -126,7 +126,7 @@ void resumeShouldDoNothingWhenStateIsNull() throws IOException { @Test void resumeShouldDoNothingWhenStateDoesNotImplementTaskAwareState() throws IOException { - var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100); + var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100, 10); var state = new SimpleState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); var stateRecord = new AggregateStateRecord(null, "SimpleState", 1, payload, @@ -235,7 +235,7 @@ void hasActiveAgentTasksShouldReturnFalseWhenStateIsNull() throws IOException { @Test void hasActiveAgentTasksShouldReturnFalseWhenNotTaskAware() throws IOException { - var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100); + var simpleRuntime = new KafkaAgenticAggregateRuntime(delegate, objectMapper, SimpleState.class, agentPlatform, aggregate, 100, 10); var state = new SimpleState("agg-1"); byte[] payload = objectMapper.writeValueAsBytes(state); var stateRecord = new AggregateStateRecord(null, "SimpleState", 1, payload, diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java index dee3b6bd..8e25e434 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java @@ -65,7 +65,17 @@ * The sliding-window capacity of the memory system. When the number of stored memories * exceeds this limit the oldest entries are evicted to make room for new ones. */ - int maxMemories() default 100; + int maxTotalMemories() default 100; + + /** + * The maximum number of net new memories that the {@code MemoryDistillerAgent} may add + * in a single distillation pass (i.e. {@code stored − revoked ≤ maxMemoriesAdded}). + * + *

    This acts as a per-distillation budget, independent of {@link #maxTotalMemories()}. + * Set to a low value to avoid overwhelming the memory store with many entries in a single + * agent run. + */ + int maxMemoriesAdded() default 10; /** * Command classes to be processed by the AI agent instead of a deterministic diff --git a/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java b/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java index f8538add..0542c0b3 100644 --- a/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java +++ b/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java @@ -66,8 +66,8 @@ public DomainEvent getCreateDomainEvent() { } } - /** Test aggregate annotated with a custom maxMemories value. */ - @AgenticAggregateInfo(value = "CustomAgentic", stateClass = TestState.class, maxMemories = 50) + /** Test aggregate annotated with custom maxTotalMemories and maxMemoriesAdded values. */ + @AgenticAggregateInfo(value = "CustomAgentic", stateClass = TestState.class, maxTotalMemories = 50, maxMemoriesAdded = 5) static class CustomMaxMemoriesAggregate implements AgenticAggregate { @Override public Class getStateClass() { @@ -84,14 +84,28 @@ public DomainEvent getCreateDomainEvent() { void defaultMaxMemoriesShouldBe100() { AgenticAggregateInfo info = DefaultAgenticAggregate.class.getAnnotation(AgenticAggregateInfo.class); assertThat(info).isNotNull(); - assertThat(info.maxMemories()).isEqualTo(100); + assertThat(info.maxTotalMemories()).isEqualTo(100); + } + + @Test + void defaultMaxMemoriesAddedShouldBe10() { + AgenticAggregateInfo info = DefaultAgenticAggregate.class.getAnnotation(AgenticAggregateInfo.class); + assertThat(info).isNotNull(); + assertThat(info.maxMemoriesAdded()).isEqualTo(10); } @Test void customMaxMemoriesShouldBeHonored() { AgenticAggregateInfo info = CustomMaxMemoriesAggregate.class.getAnnotation(AgenticAggregateInfo.class); assertThat(info).isNotNull(); - assertThat(info.maxMemories()).isEqualTo(50); + assertThat(info.maxTotalMemories()).isEqualTo(50); + } + + @Test + void customMaxMemoriesAddedShouldBeHonored() { + AgenticAggregateInfo info = CustomMaxMemoriesAggregate.class.getAnnotation(AgenticAggregateInfo.class); + assertThat(info).isNotNull(); + assertThat(info.maxMemoriesAdded()).isEqualTo(5); } @Test diff --git a/main/eventcatalog/src/test/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessorTest.java b/main/eventcatalog/src/test/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessorTest.java index 86222210..8892542c 100644 --- a/main/eventcatalog/src/test/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessorTest.java +++ b/main/eventcatalog/src/test/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessorTest.java @@ -1000,7 +1000,7 @@ public AssistantState withoutMemory(String memoryId) { value = "Assistant", stateClass = AssistantState.class, description = "AI Assistant AgenticAggregate", - maxMemories = 50) + maxTotalMemories = 50) @SuppressWarnings("unused") public final class Assistant implements AgenticAggregate { @Override From eb1c8c26d34e8d5c84369e4aed840bb19ea3c917 Mon Sep 17 00:00:00 2001 From: Joost van de Wijgerd Date: Sat, 11 Apr 2026 14:16:29 +0200 Subject: [PATCH 04/10] Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../akces/agentic/runtime/KafkaAgenticAggregateRuntime.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index d2fb923a..d13c70f7 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -105,6 +105,9 @@ public KafkaAgenticAggregateRuntime(KafkaAggregateRuntime delegate, this.stateClass = Objects.requireNonNull(stateClass, "stateClass must not be null"); this.agentPlatform = Objects.requireNonNull(agentPlatform, "agentPlatform must not be null"); this.aggregate = Objects.requireNonNull(aggregate, "aggregate must not be null"); + if (maxMemories < 0) { + throw new IllegalArgumentException("maxMemories must be greater than or equal to 0"); + } this.maxTotalMemories = maxTotalMemories; this.maxMemoriesAdded = maxMemoriesAdded; } From 3103bbcef6be9c1ad6348a9a5dc06f9a9245ebc7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 12:23:21 +0000 Subject: [PATCH 05/10] Address review: GOAP planner, ObjectMapper injection, remove explicit return format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Change MemoryDistillerAgent planner from UTILITY to GOAP (has @AchievesGoal) - Inject ObjectMapper via constructor and use it for JSON serialization of history, blackboard objects, and existing memories in the prompt - Remove explicit return format instructions from prompt since .createObject() already instructs the LLM on the return schema - Fix maxMemories → maxTotalMemories bug in constructor validation - Add maxMemoriesAdded >= 0 validation Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/c67ccc3c-4bc2-4df9-84d7-9a3c6fbc9a46 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../agentic/embabel/MemoryDistillerAgent.java | 44 ++++++++++++------- .../runtime/KafkaAgenticAggregateRuntime.java | 7 ++- 2 files changed, 34 insertions(+), 17 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 60efc065..8516cddc 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -22,6 +22,7 @@ import com.embabel.agent.api.annotation.Agent; import com.embabel.agent.api.common.ActionContext; import com.embabel.agent.api.common.PlannerType; +import tools.jackson.databind.ObjectMapper; import java.util.List; @@ -46,7 +47,7 @@ */ @Agent(name = MemoryDistillerAgent.AGENT_NAME, description = "Distills relevant memories from a completed agent process", - planner = PlannerType.UTILITY) + planner = PlannerType.GOAP) public class MemoryDistillerAgent { /** @@ -55,6 +56,18 @@ public class MemoryDistillerAgent { */ public static final String AGENT_NAME = "MemoryDistiller"; + private final ObjectMapper objectMapper; + + /** + * Creates a new {@code MemoryDistillerAgent}. + * + * @param objectMapper the Jackson {@link ObjectMapper} used for JSON serialization + * of blackboard objects and memories in the LLM prompt + */ + public MemoryDistillerAgent(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + /** * Distills memories from a completed agent process by analyzing the process * history and blackboard contents using an LLM. @@ -128,11 +141,6 @@ memories should be revoked (because they are outdated, incorrect, or superseded) .append(maxNewMemories).append("\n"); sb.append(""" - - Each stored memory must have: agenticAggregateId, memoryId (UUID), \ - subject (1-2 words), fact (max 200 chars), citations (source reference), \ - reason (why it should be stored), and storedAt (ISO-8601 instant) - - Each revoked memory must have: agenticAggregateId, memoryId (from existing \ - memories), reason, and revokedAt (ISO-8601 instant) - Only store memories that are actionable, likely to remain relevant, and \ cannot always be inferred from limited context - Only revoke memories that are clearly outdated, incorrect, or superseded \ @@ -143,7 +151,7 @@ memories should be revoked (because they are outdated, incorrect, or superseded) sb.append("PROCESS HISTORY (actions executed):\n"); if (history != null && !history.isEmpty()) { for (Object invocation : history) { - sb.append("- ").append(invocation).append("\n"); + sb.append("- ").append(serialize(invocation)).append("\n"); } } else { sb.append("(no actions recorded)\n"); @@ -153,7 +161,7 @@ memories should be revoked (because they are outdated, incorrect, or superseded) if (blackboardObjects != null && !blackboardObjects.isEmpty()) { for (Object obj : blackboardObjects) { sb.append("- [").append(obj.getClass().getSimpleName()).append("] ") - .append(obj).append("\n"); + .append(serialize(obj)).append("\n"); } } else { sb.append("(no objects)\n"); @@ -162,18 +170,24 @@ memories should be revoked (because they are outdated, incorrect, or superseded) sb.append("\nEXISTING MEMORIES:\n"); if (existingMemories != null && !existingMemories.isEmpty()) { for (Object memory : existingMemories) { - sb.append("- ").append(memory).append("\n"); + sb.append("- ").append(serialize(memory)).append("\n"); } } else { sb.append("(no existing memories)\n"); } - sb.append(""" - - Return a JSON object with 'stored' (list of MemoryStoredEvent instances) \ - and 'revoked' (list of MemoryRevokedEvent instances). \ - If no memories should be stored or revoked, return empty lists."""); - return sb.toString(); } + + /** + * Serializes an object to JSON using the injected {@link ObjectMapper}. + * Falls back to {@link Object#toString()} if serialization fails. + */ + private String serialize(Object obj) { + try { + return objectMapper.writeValueAsString(obj); + } catch (Exception e) { + return obj.toString(); + } + } } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index d13c70f7..f0b2cfd5 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -105,8 +105,11 @@ public KafkaAgenticAggregateRuntime(KafkaAggregateRuntime delegate, this.stateClass = Objects.requireNonNull(stateClass, "stateClass must not be null"); this.agentPlatform = Objects.requireNonNull(agentPlatform, "agentPlatform must not be null"); this.aggregate = Objects.requireNonNull(aggregate, "aggregate must not be null"); - if (maxMemories < 0) { - throw new IllegalArgumentException("maxMemories must be greater than or equal to 0"); + if (maxTotalMemories < 0) { + throw new IllegalArgumentException("maxTotalMemories must be >= 0"); + } + if (maxMemoriesAdded < 0) { + throw new IllegalArgumentException("maxMemoriesAdded must be >= 0"); } this.maxTotalMemories = maxTotalMemories; this.maxMemoriesAdded = maxMemoriesAdded; From 5bcc8c810ad1de4c5b848fe98fa7254f5f3147c3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 12:26:45 +0000 Subject: [PATCH 06/10] Add null check for ObjectMapper in MemoryDistillerAgent constructor Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/c67ccc3c-4bc2-4df9-84d7-9a3c6fbc9a46 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akces/agentic/embabel/MemoryDistillerAgent.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 8516cddc..7faa28a6 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -25,6 +25,7 @@ import tools.jackson.databind.ObjectMapper; import java.util.List; +import java.util.Objects; /** * Embabel agent responsible for distilling relevant memories from a successfully @@ -65,7 +66,7 @@ public class MemoryDistillerAgent { * of blackboard objects and memories in the LLM prompt */ public MemoryDistillerAgent(ObjectMapper objectMapper) { - this.objectMapper = objectMapper; + this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); } /** From 3d6dda06510bfdac5009455ed4da0157e6074128 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 14:10:54 +0000 Subject: [PATCH 07/10] Use JSON markdown code fencing in prompt, change defaults to maxTotalMemories=50, maxMemoriesAdded=5 Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/7180951b-dc17-4c5f-89fe-515a51c44b07 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../agentic/embabel/MemoryDistillerAgent.java | 40 +++++++++++++------ .../annotations/AgenticAggregateInfo.java | 4 +- 2 files changed, 29 insertions(+), 15 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 7faa28a6..04016eed 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -149,33 +149,47 @@ memories should be revoked (because they are outdated, incorrect, or superseded) """); - sb.append("PROCESS HISTORY (actions executed):\n"); + sb.append("PROCESS HISTORY (actions executed):\n```json\n"); if (history != null && !history.isEmpty()) { - for (Object invocation : history) { - sb.append("- ").append(serialize(invocation)).append("\n"); + sb.append("[\n"); + for (int i = 0; i < history.size(); i++) { + sb.append(serialize(history.get(i))); + if (i < history.size() - 1) sb.append(","); + sb.append("\n"); } + sb.append("]\n"); } else { - sb.append("(no actions recorded)\n"); + sb.append("[]\n"); } + sb.append("```\n"); - sb.append("\nBLACKBOARD OBJECTS (process context and results):\n"); + sb.append("\nBLACKBOARD OBJECTS (process context and results):\n```json\n"); if (blackboardObjects != null && !blackboardObjects.isEmpty()) { - for (Object obj : blackboardObjects) { - sb.append("- [").append(obj.getClass().getSimpleName()).append("] ") - .append(serialize(obj)).append("\n"); + sb.append("[\n"); + for (int i = 0; i < blackboardObjects.size(); i++) { + sb.append(serialize(blackboardObjects.get(i))); + if (i < blackboardObjects.size() - 1) sb.append(","); + sb.append("\n"); } + sb.append("]\n"); } else { - sb.append("(no objects)\n"); + sb.append("[]\n"); } + sb.append("```\n"); - sb.append("\nEXISTING MEMORIES:\n"); + sb.append("\nEXISTING MEMORIES:\n```json\n"); if (existingMemories != null && !existingMemories.isEmpty()) { - for (Object memory : existingMemories) { - sb.append("- ").append(serialize(memory)).append("\n"); + sb.append("[\n"); + for (int i = 0; i < existingMemories.size(); i++) { + sb.append(serialize(existingMemories.get(i))); + if (i < existingMemories.size() - 1) sb.append(","); + sb.append("\n"); } + sb.append("]\n"); } else { - sb.append("(no existing memories)\n"); + sb.append("[]\n"); } + sb.append("```\n"); return sb.toString(); } diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java index 8e25e434..252ffe40 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java @@ -65,7 +65,7 @@ * The sliding-window capacity of the memory system. When the number of stored memories * exceeds this limit the oldest entries are evicted to make room for new ones. */ - int maxTotalMemories() default 100; + int maxTotalMemories() default 50; /** * The maximum number of net new memories that the {@code MemoryDistillerAgent} may add @@ -75,7 +75,7 @@ * Set to a low value to avoid overwhelming the memory store with many entries in a single * agent run. */ - int maxMemoriesAdded() default 10; + int maxMemoriesAdded() default 5; /** * Command classes to be processed by the AI agent instead of a deterministic From c8ec319d47e735c024a58e8090b789df2ccc6ff5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 14:22:13 +0000 Subject: [PATCH 08/10] Fix test failures and add agentTask to distiller blackboard bindings Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/2e093a9b-24ea-4985-8829-5188dc97827f Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../agentic/embabel/MemoryDistillerAgent.java | 18 +++++++++++++++--- .../runtime/KafkaAgenticAggregateRuntime.java | 7 +++++-- .../runtime/MemoryDistillationTest.java | 2 ++ .../aggregate/AgenticAggregateInfoTest.java | 8 ++++---- 4 files changed, 26 insertions(+), 9 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 04016eed..715c27d0 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -75,6 +75,8 @@ public MemoryDistillerAgent(ObjectMapper objectMapper) { * *

    The method reads the following bindings from the blackboard: *

      + *
    • {@code "agentTask"} — the {@link org.elasticsoftware.akces.aggregate.AssignedTask} + * that triggered the completed process
      • *
    • {@code "history"} — the execution history from the completed process
      • *
    • {@code "blackboardObjects"} — all objects from the completed process's * blackboard
      • @@ -92,6 +94,7 @@ public MemoryDistillerAgent(ObjectMapper objectMapper) { public MemoryDistillationResult distillMemories(ActionContext context) { var blackboard = context.getProcessContext().getAgentProcess().getBlackboard(); + Object agentTaskBinding = blackboard.get("agentTask"); Object historyBinding = blackboard.get("history"); List history = historyBinding instanceof List historyList ? historyList @@ -117,14 +120,15 @@ public MemoryDistillationResult distillMemories(ActionContext context) { int capacityLeft = Math.max(0, maxTotalMemories - currentCount); int effectiveLimit = Math.min(capacityLeft, maxMemoriesAdded); - String prompt = buildPrompt(history, blackboardObjects, existingMemories, effectiveLimit); + String prompt = buildPrompt(agentTaskBinding, history, blackboardObjects, existingMemories, effectiveLimit); return context.ai() .withDefaultLlm() .createObject(prompt, MemoryDistillationResult.class); } - private String buildPrompt(List history, + private String buildPrompt(Object agentTask, + List history, List blackboardObjects, List existingMemories, int maxNewMemories) { @@ -149,7 +153,15 @@ memories should be revoked (because they are outdated, incorrect, or superseded) """); - sb.append("PROCESS HISTORY (actions executed):\n```json\n"); + sb.append("AGENT TASK:\n```json\n"); + if (agentTask != null) { + sb.append(serialize(agentTask)).append("\n"); + } else { + sb.append("null\n"); + } + sb.append("```\n"); + + sb.append("\nPROCESS HISTORY (actions executed):\n```json\n"); if (history != null && !history.isEmpty()) { sb.append("[\n"); for (int i = 0; i < history.size(); i++) { diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index f0b2cfd5..7dafc573 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -530,7 +530,7 @@ public void resumeNextAgentTask(Consumer protocolRecordConsumer, // Distill memories from successfully completed processes if (statusCode == AgentProcessStatusCode.COMPLETED) { - List memoryEvents = distillMemories(agentProcess, state); + List memoryEvents = distillMemories(agentProcess, state, task); if (!memoryEvents.isEmpty()) { tickEvents = Stream.concat(tickEvents, memoryEvents.stream()); } @@ -559,10 +559,12 @@ public void resumeNextAgentTask(Consumer protocolRecordConsumer, * * @param completedProcess the agent process that has completed successfully * @param state the current aggregate state + * @param task the assigned task that triggered the agent process * @return a list of {@link MemoryStoredEvent} and {@link MemoryRevokedEvent} instances; * may be empty if no memories need to be changed */ - private List distillMemories(AgentProcess completedProcess, AggregateState state) { + private List distillMemories(AgentProcess completedProcess, AggregateState state, + AssignedTask task) { Agent memoryDistillerAgent = resolveMemoryDistillerAgent(); if (memoryDistillerAgent == null) { logger.warn("MemoryDistillerAgent not deployed on the platform; skipping memory distillation"); @@ -574,6 +576,7 @@ private List distillMemories(AgentProcess completedProcess, Aggrega : List.of(); Map bindings = new LinkedHashMap<>(); + bindings.put("agentTask", task); bindings.put("history", completedProcess.getHistory()); bindings.put("blackboardObjects", completedProcess.getBlackboard().getObjects()); bindings.put("existingMemories", currentMemories); diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java index 8fe54ddd..9c15f3fe 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java @@ -639,12 +639,14 @@ void distillationShouldPassCorrectBindingsToDistillerProcess() throws IOExceptio eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), bindingsCaptor.capture()); Map bindings = bindingsCaptor.getValue(); + assertThat(bindings).containsKey("agentTask"); assertThat(bindings).containsKey("history"); assertThat(bindings).containsKey("blackboardObjects"); assertThat(bindings).containsKey("existingMemories"); assertThat(bindings).containsKey("maxTotalMemories"); assertThat(bindings).containsKey("maxMemoriesAdded"); + assertThat(bindings.get("agentTask")).isEqualTo(task); assertThat(bindings.get("existingMemories")).isEqualTo(List.of(existingMemory)); // maxTotalMemories=100, maxMemoriesAdded=10 assertThat(bindings.get("maxTotalMemories")).isEqualTo(100); diff --git a/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java b/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java index 0542c0b3..bd63f18b 100644 --- a/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java +++ b/main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java @@ -81,17 +81,17 @@ public DomainEvent getCreateDomainEvent() { } @Test - void defaultMaxMemoriesShouldBe100() { + void defaultMaxMemoriesShouldBe50() { AgenticAggregateInfo info = DefaultAgenticAggregate.class.getAnnotation(AgenticAggregateInfo.class); assertThat(info).isNotNull(); - assertThat(info.maxTotalMemories()).isEqualTo(100); + assertThat(info.maxTotalMemories()).isEqualTo(50); } @Test - void defaultMaxMemoriesAddedShouldBe10() { + void defaultMaxMemoriesAddedShouldBe5() { AgenticAggregateInfo info = DefaultAgenticAggregate.class.getAnnotation(AgenticAggregateInfo.class); assertThat(info).isNotNull(); - assertThat(info.maxMemoriesAdded()).isEqualTo(10); + assertThat(info.maxMemoriesAdded()).isEqualTo(5); } @Test From 6358a9192f3be7a90de42652c41b4780e0393bde Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 14:41:46 +0000 Subject: [PATCH 09/10] Refactor: use strongly-typed MemoryDistillationInput for action method params Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/a03c329c-ac6e-4e7c-8779-1b7a390b1f5f Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../embabel/MemoryDistillationInput.java | 48 ++++++++++++++++ .../agentic/embabel/MemoryDistillerAgent.java | 57 ++++++------------- .../runtime/KafkaAgenticAggregateRuntime.java | 23 +++++--- .../runtime/MemoryDistillationTest.java | 23 ++++---- 4 files changed, 90 insertions(+), 61 deletions(-) create mode 100644 main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationInput.java diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationInput.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationInput.java new file mode 100644 index 00000000..d132c461 --- /dev/null +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillationInput.java @@ -0,0 +1,48 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.elasticsoftware.akces.agentic.embabel; + +import com.embabel.agent.core.ActionInvocation; +import org.elasticsoftware.akces.aggregate.AgenticAggregateMemory; +import org.elasticsoftware.akces.aggregate.AssignedTask; + +import java.util.List; + +/** + * Strongly-typed input for the {@link MemoryDistillerAgent} action method. + * + *

      This record is placed on the Embabel Blackboard by the Akces runtime + * and injected into the agent's action method by the Embabel framework + * via type-based parameter resolution. + * + * @param agentTask the assigned task that triggered the completed process + * @param history the execution history (action invocations) from the completed process + * @param blackboardObjects all objects from the completed process's blackboard + * @param existingMemories the current list of memories from the aggregate state + * @param maxTotalMemories the total memory capacity of the aggregate + * @param maxMemoriesAdded the per-distillation budget for net new memories + */ +public record MemoryDistillationInput( + AssignedTask agentTask, + List history, + List blackboardObjects, + List existingMemories, + int maxTotalMemories, + int maxMemoriesAdded +) { +} diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java index 715c27d0..bc759eb6 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/MemoryDistillerAgent.java @@ -73,54 +73,31 @@ public MemoryDistillerAgent(ObjectMapper objectMapper) { * Distills memories from a completed agent process by analyzing the process * history and blackboard contents using an LLM. * - *

      The method reads the following bindings from the blackboard: + *

      The Embabel framework injects the {@link MemoryDistillationInput} parameter + * from the Blackboard via type-based resolution. The input contains: *

        - *
      • {@code "agentTask"} — the {@link org.elasticsoftware.akces.aggregate.AssignedTask} + *
      • {@code agentTask} — the {@link org.elasticsoftware.akces.aggregate.AssignedTask} * that triggered the completed process
        • - *
      • {@code "history"} — the execution history from the completed process
        • - *
      • {@code "blackboardObjects"} — all objects from the completed process's - * blackboard
        • - *
      • {@code "existingMemories"} — the current list of memories from the aggregate - * state
        • - *
      • {@code "maxTotalMemories"} — the total memory capacity of the aggregate
        • - *
      • {@code "maxMemoriesAdded"} — the per-distillation budget for net new memories
        • + *
      • {@code history} — the execution history from the completed process
        • + *
      • {@code blackboardObjects} — all objects from the completed process's blackboard
        • + *
      • {@code existingMemories} — the current list of memories from the aggregate state
        • + *
      • {@code maxTotalMemories} — the total memory capacity of the aggregate
        • + *
      • {@code maxMemoriesAdded} — the per-distillation budget for net new memories
        • *
      * - * @param context the action context providing access to AI capabilities and blackboard + * @param input the strongly-typed distillation input, injected by the Embabel framework + * @param context the action context providing access to AI capabilities * @return a {@link MemoryDistillationResult} with memories to store and revoke */ @Action(description = "Distill relevant memories from a completed agent process") @AchievesGoal(description = "Distill and manage memories from completed agent processes") - public MemoryDistillationResult distillMemories(ActionContext context) { - var blackboard = context.getProcessContext().getAgentProcess().getBlackboard(); - - Object agentTaskBinding = blackboard.get("agentTask"); - Object historyBinding = blackboard.get("history"); - List history = historyBinding instanceof List historyList - ? historyList - : List.of(); - Object blackboardObjectsBinding = blackboard.get("blackboardObjects"); - List blackboardObjects = blackboardObjectsBinding instanceof List objectList - ? objectList - : List.of(); - Object existingMemoriesBinding = blackboard.get("existingMemories"); - List existingMemories = existingMemoriesBinding instanceof List memoryList - ? memoryList - : List.of(); - Object maxTotalMemoriesBinding = blackboard.get("maxTotalMemories"); - int maxTotalMemories = maxTotalMemoriesBinding instanceof Number number - ? number.intValue() - : 0; - Object maxMemoriesAddedBinding = blackboard.get("maxMemoriesAdded"); - int maxMemoriesAdded = maxMemoriesAddedBinding instanceof Number number - ? number.intValue() - : 0; - - int currentCount = existingMemories.size(); - int capacityLeft = Math.max(0, maxTotalMemories - currentCount); - int effectiveLimit = Math.min(capacityLeft, maxMemoriesAdded); - - String prompt = buildPrompt(agentTaskBinding, history, blackboardObjects, existingMemories, effectiveLimit); + public MemoryDistillationResult distillMemories(MemoryDistillationInput input, ActionContext context) { + int currentCount = input.existingMemories().size(); + int capacityLeft = Math.max(0, input.maxTotalMemories() - currentCount); + int effectiveLimit = Math.min(capacityLeft, input.maxMemoriesAdded()); + + String prompt = buildPrompt(input.agentTask(), input.history(), input.blackboardObjects(), + input.existingMemories(), effectiveLimit); return context.ai() .withDefaultLlm() diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index 7dafc573..01febc3b 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -25,6 +25,7 @@ import jakarta.annotation.Nullable; import org.apache.kafka.common.errors.SerializationException; import org.elasticsoftware.akces.agentic.AgenticAggregateRuntime; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationInput; import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationResult; import org.elasticsoftware.akces.agentic.embabel.MemoryDistillerAgent; import org.elasticsoftware.akces.agentic.events.AgentTaskAssignedEvent; @@ -549,9 +550,10 @@ public void resumeNextAgentTask(Consumer protocolRecordConsumer, * by running the {@link MemoryDistillerAgent} to completion. * *

      Creates a separate agent process for the {@link MemoryDistillerAgent}, populates - * its blackboard with the completed process's history and blackboard objects, the - * current memories, and the maximum number of net new memories allowed, then runs - * the process to completion via {@link AgentProcess#run()}. + * its blackboard with a single {@link MemoryDistillationInput} containing the completed + * process's history, blackboard objects, current memories, and memory capacity constraints. + * The Embabel framework injects this input into the agent's action method via type-based + * parameter resolution. The process is then run to completion via {@link AgentProcess#run()}. * *

      The net memory constraint ensures that * {@code storedCount - revokedCount <= maxMemories - currentMemoryCount}, preventing @@ -575,13 +577,16 @@ private List distillMemories(AgentProcess completedProcess, Aggrega ? mas.getMemories() : List.of(); + MemoryDistillationInput distillationInput = new MemoryDistillationInput( + task, + completedProcess.getHistory(), + completedProcess.getBlackboard().getObjects(), + currentMemories, + maxTotalMemories, + maxMemoriesAdded); + Map bindings = new LinkedHashMap<>(); - bindings.put("agentTask", task); - bindings.put("history", completedProcess.getHistory()); - bindings.put("blackboardObjects", completedProcess.getBlackboard().getObjects()); - bindings.put("existingMemories", currentMemories); - bindings.put("maxTotalMemories", maxTotalMemories); - bindings.put("maxMemoriesAdded", maxMemoriesAdded); + bindings.put("input", distillationInput); try { AgentProcess distillerProcess = agentPlatform.createAgentProcess( diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java index 9c15f3fe..0a6fe129 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/MemoryDistillationTest.java @@ -23,6 +23,7 @@ import com.embabel.agent.core.AgentProcessStatusCode; import com.embabel.agent.core.Blackboard; import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationInput; import org.elasticsoftware.akces.agentic.embabel.MemoryDistillationResult; import org.elasticsoftware.akces.agentic.embabel.MemoryDistillerAgent; import org.elasticsoftware.akces.agentic.events.AgentTaskFinishedEvent; @@ -639,17 +640,15 @@ void distillationShouldPassCorrectBindingsToDistillerProcess() throws IOExceptio eq(memoryDistillerAgent), eq(ProcessOptions.DEFAULT), bindingsCaptor.capture()); Map bindings = bindingsCaptor.getValue(); - assertThat(bindings).containsKey("agentTask"); - assertThat(bindings).containsKey("history"); - assertThat(bindings).containsKey("blackboardObjects"); - assertThat(bindings).containsKey("existingMemories"); - assertThat(bindings).containsKey("maxTotalMemories"); - assertThat(bindings).containsKey("maxMemoriesAdded"); - - assertThat(bindings.get("agentTask")).isEqualTo(task); - assertThat(bindings.get("existingMemories")).isEqualTo(List.of(existingMemory)); - // maxTotalMemories=100, maxMemoriesAdded=10 - assertThat(bindings.get("maxTotalMemories")).isEqualTo(100); - assertThat(bindings.get("maxMemoriesAdded")).isEqualTo(10); + assertThat(bindings).containsKey("input"); + assertThat(bindings.get("input")).isInstanceOf(MemoryDistillationInput.class); + + MemoryDistillationInput input = (MemoryDistillationInput) bindings.get("input"); + assertThat(input.agentTask()).isEqualTo(task); + assertThat(input.history()).isEqualTo(List.of()); + assertThat(input.blackboardObjects()).isEqualTo(List.of("some-object")); + assertThat(input.existingMemories()).isEqualTo(List.of(existingMemory)); + assertThat(input.maxTotalMemories()).isEqualTo(100); + assertThat(input.maxMemoriesAdded()).isEqualTo(10); } } From ee51f3f4e0d73e70ec979de97873042c7e3887c2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Apr 2026 15:00:34 +0000 Subject: [PATCH 10/10] Remove unused aggregateId param from translateDistillationResult Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/0d443301-e294-4183-adf3-3c217d7c5007 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akces/agentic/runtime/KafkaAgenticAggregateRuntime.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java index 01febc3b..2edc1adc 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java @@ -603,7 +603,7 @@ private List distillMemories(AgentProcess completedProcess, Aggrega int capacityLeft = Math.max(0, maxTotalMemories - currentMemories.size()); int effectiveLimit = Math.min(capacityLeft, maxMemoriesAdded); - return translateDistillationResult(result, state.getAggregateId(), currentMemories, effectiveLimit); + return translateDistillationResult(result, currentMemories, effectiveLimit); } catch (Exception e) { logger.warn("Memory distillation failed for aggregate {}; proceeding without memory updates", getName(), e); @@ -633,13 +633,11 @@ private Agent resolveMemoryDistillerAgent() { * enforced by truncating stored memories when the limit would be exceeded. * * @param result the distillation result from the agent - * @param aggregateId the aggregate identifier for the events * @param currentMemories the current memories from the aggregate state * @param maxNewMemories the maximum number of net new memories allowed * @return an unmodifiable list of domain events */ private List translateDistillationResult(MemoryDistillationResult result, - String aggregateId, List currentMemories, int maxNewMemories) { List events = new ArrayList<>();