From 4364638c5917eb02ae6160ee1bfc4ae3b6740a25 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 10:42:38 +0000 Subject: [PATCH 01/10] Add plan for agent resolution refactoring Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/e6ed61c0-c5ac-4bbc-9cab-5e7276244968 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- plans/agent-resolution-refactoring.md | 89 +++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 plans/agent-resolution-refactoring.md diff --git a/plans/agent-resolution-refactoring.md b/plans/agent-resolution-refactoring.md new file mode 100644 index 00000000..bae40cad --- /dev/null +++ b/plans/agent-resolution-refactoring.md @@ -0,0 +1,89 @@ +# Agent Resolution Refactoring Plan + +## Summary + +Refactor `AgenticCommandHandlerFunctionAdapter` and `AgenticEventHandlerFunctionAdapter` to not +inject the `Agent` instance directly. Instead, infer the agent at runtime based on the +`AgenticAggregate` name by searching `agentPlatform.agents()`. If no matching agent is found, +fall back to using `Autonomy` via `agentPlatform.getPlatformServices().autonomy()` to let Embabel +decide on the best solution. + +## Motivation + +The current approach requires a 1:1 Spring bean naming convention (`{aggregateName}Agent`) and +fails fast at startup if the bean is missing. The new approach: + +1. **Decouples** agent registration from the Spring bean lifecycle — agents deployed to the + `AgentPlatform` at any time will be discovered automatically. +2. **Enables fallback** via Embabel's `Autonomy` when no agent is explicitly named after the + aggregate, letting the platform choose the best agent/goal combination. +3. **Simplifies** the `AgenticAggregateRuntimeFactory` by removing eager agent resolution. + +## Phase 1: Adapter Refactoring + +### 1.1 AgenticCommandHandlerFunctionAdapter + +- Replace the `Agent agent` constructor parameter and field with `String aggregateName`. +- In `apply()`, resolve the agent lazily: + 1. Search `agentPlatform.agents()` for an `Agent` whose `getName()` matches the aggregate name + (exact match or `{aggregateName}Agent` suffix match). + 2. If found → use existing `agentPlatform.createAgentProcess(agent, ProcessOptions.DEFAULT, bindings)` + with the manual tick loop. + 3. If not found → fall back to `agentPlatform.getPlatformServices().autonomy() + .chooseAndAccomplishGoal(GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings)`. + Extract the `AgentProcess` from the returned `AgentProcessExecution`. +- Collect events from the blackboard identically in both paths. + +### 1.2 AgenticEventHandlerFunctionAdapter + +- Same changes as 1.1 but for the event handling path. +- The `event` binding (instead of `command`) is placed on the blackboard. + +### 1.3 Agent Resolution Helper + +- Extract the agent-name-matching logic into a private helper method shared by both adapters + (or a static utility) to avoid duplication. The method signature: + ```java + Optional resolveAgentByName(AgentPlatform platform, String aggregateName) + ``` + Matching rules: + 1. Exact match: `agent.getName().equals(aggregateName)` + 2. Agent-suffix match: `agent.getName().equals(aggregateName + "Agent")` + +## Phase 2: Factory Cleanup + +### 2.1 AgenticAggregateRuntimeFactory + +- Remove the `resolveAgent(String aggregateName)` method. +- Remove the `Agent agent` local variable and the lazy resolution logic in `createRuntime()`. +- Update `processAgentHandledCommand()` and `processAgentHandledEvent()` to pass the aggregate + name (String) instead of an `Agent` to the adapter constructors. + +## Phase 3: Testing + +### 3.1 Unit Tests + +- Add tests for `AgenticCommandHandlerFunctionAdapter`: + - Agent found by exact name match → process created via `createAgentProcess`. + - Agent found by suffix match → process created via `createAgentProcess`. + - No agent found → Autonomy fallback invoked. +- Add tests for `AgenticEventHandlerFunctionAdapter`: + - Same three scenarios. +- Verify that the Autonomy fallback correctly collects events from the resulting blackboard. + +### 3.2 Existing Test Validation + +- Run the full agentic module test suite (`mvn test -pl main/agentic`) to ensure no regressions. + +## Dependencies + +- Embabel `embabel-agent-api` 0.3.5 — `Autonomy`, `GoalChoiceApprover`, `AgentProcessExecution`. +- No new library dependencies required. + +## Risk Assessment + +- **Low risk**: The matching logic is straightforward and well-tested. +- **Medium risk**: The `Autonomy.chooseAndAccomplishGoal()` fallback runs the process to completion + synchronously. If Embabel's internal timeout/budget is not aligned with Kafka consumer polling + intervals, the consumer may be considered dead. Mitigation: use `ProcessOptions` with an + appropriate budget in a future iteration. From 69efe1d6b232667f99386c09b82586cfe94f3dba Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 10:47:44 +0000 Subject: [PATCH 02/10] Refactor adapters to infer Agent by name with Autonomy fallback Replace direct Agent injection in AgenticCommandHandlerFunctionAdapter and AgenticEventHandlerFunctionAdapter with runtime agent resolution based on the aggregate name. If no matching agent is found, fall back to Autonomy.chooseAndAccomplishGoal() to let Embabel decide. Remove resolveAgent() from AgenticAggregateRuntimeFactory; pass aggregate name (String) instead of Agent to adapter constructors. Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/e6ed61c0-c5ac-4bbc-9cab-5e7276244968 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../beans/AgenticAggregateRuntimeFactory.java | 46 +----- .../AgenticCommandHandlerFunctionAdapter.java | 149 +++++++++++++----- .../AgenticEventHandlerFunctionAdapter.java | 122 +++++++++----- 3 files changed, 196 insertions(+), 121 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 d3a90693..34ba40d1 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 @@ -17,7 +17,6 @@ package org.elasticsoftware.akces.agentic.beans; -import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import org.elasticsoftware.akces.agentic.AgenticAggregateRuntime; import org.elasticsoftware.akces.agentic.runtime.AgenticCommandHandlerFunctionAdapter; @@ -158,30 +157,6 @@ private AgentPlatform resolveAgentPlatform(String aggregateName) { } } - /** - * Resolves the {@link Agent} for this aggregate from the {@link ApplicationContext}. - * - *

Looks for a Spring bean of type {@link Agent} named {@code {aggregateName}Agent}. - * The implementing application is responsible for registering this bean. If no such - * bean is found, a fatal error is raised. - * - * @param aggregateName the aggregate name used to derive the agent bean name - * @return the resolved {@link Agent}; never {@code null} - * @throws IllegalStateException if no matching {@link Agent} bean is found - */ - private Agent resolveAgent(String aggregateName) { - String agentBeanName = aggregateName + "Agent"; - try { - return applicationContext.getBean(agentBeanName, Agent.class); - } catch (BeansException e) { - throw new IllegalStateException( - "No Agent bean found with name '" + agentBeanName + "' for agentic aggregate '" - + aggregateName + "'. " - + "The implementing application must provide a Spring bean of type " - + "com.embabel.agent.core.Agent named '" + agentBeanName + "'.", e); - } - } - @SuppressWarnings("unchecked") private KafkaAggregateRuntime createRuntime(AgenticAggregateInfo agenticInfo, AgenticAggregate aggregate, @@ -282,22 +257,17 @@ private KafkaAggregateRuntime createRuntime(AgenticAggregateInfo agenticInfo, buildAgentProducedErrorTypes(agenticInfo.agentProducedErrors()); agentProducedErrorTypes.forEach(runtimeBuilder::addDomainEvent); - // Lazy Agent resolution: only needed when agent-handled commands or events exist. - boolean hasAgentHandlers = agenticInfo.agentHandledCommands().length > 0 - || agenticInfo.agentHandledEvents().length > 0; - Agent agent = hasAgentHandlers ? resolveAgent(agenticInfo.value()) : null; - // Process agentHandledCommands — register AgenticCommandHandlerFunctionAdapter for (Class commandClass : agenticInfo.agentHandledCommands()) { processAgentHandledCommand( - commandClass, aggregate, agentPlatform, agent, + commandClass, aggregate, agenticInfo.value(), agentPlatform, agentProducedErrorTypes, runtimeBuilder); } // Process agentHandledEvents — register AgenticEventHandlerFunctionAdapter for (Class eventClass : agenticInfo.agentHandledEvents()) { processAgentHandledEvent( - eventClass, aggregate, agentPlatform, agent, + eventClass, aggregate, agenticInfo.value(), agentPlatform, agentProducedErrorTypes, runtimeBuilder); } @@ -339,8 +309,8 @@ private List> buildAgentProducedErrorTypes( * * @param commandClass the command class to register * @param aggregate the owning aggregate instance + * @param aggregateName the aggregate name used for agent inference at runtime * @param agentPlatform the Embabel platform - * @param agent the deployed {@link Agent} for this aggregate * @param agentProducedErrors the error types the agent may produce * @param runtimeBuilder the runtime builder to register with */ @@ -348,8 +318,8 @@ private List> buildAgentProducedErrorTypes( private void processAgentHandledCommand( Class commandClass, AgenticAggregate aggregate, + String aggregateName, AgentPlatform agentPlatform, - Agent agent, List> agentProducedErrors, KafkaAggregateRuntime.Builder runtimeBuilder) { @@ -373,9 +343,9 @@ private void processAgentHandledCommand( AgenticCommandHandlerFunctionAdapter adapter = new AgenticCommandHandlerFunctionAdapter<>( (AgenticAggregate) aggregate, + aggregateName, commandType, agentPlatform, - agent, (List) List.of(), // producedDomainEventTypes: empty — events are registered via EventSourcingHandler adapters (List) errorTypes, // TODO: wire aggregateServicesSupplier from AkcesAgenticAggregateController (Phase 3) @@ -395,8 +365,8 @@ private void processAgentHandledCommand( * * @param eventClass the external domain event class to register * @param aggregate the owning aggregate instance + * @param aggregateName the aggregate name used for agent inference at runtime * @param agentPlatform the Embabel platform - * @param agent the deployed {@link Agent} for this aggregate * @param agentProducedErrors the error types the agent may produce * @param runtimeBuilder the runtime builder to register with */ @@ -404,8 +374,8 @@ private void processAgentHandledCommand( private void processAgentHandledEvent( Class eventClass, AgenticAggregate aggregate, + String aggregateName, AgentPlatform agentPlatform, - Agent agent, List> agentProducedErrors, KafkaAggregateRuntime.Builder runtimeBuilder) { @@ -436,9 +406,9 @@ private void processAgentHandledEvent( AgenticEventHandlerFunctionAdapter adapter = new AgenticEventHandlerFunctionAdapter<>( (AgenticAggregate) aggregate, + aggregateName, eventType, agentPlatform, - agent, (List) List.of(), // producedDomainEventTypes: empty — events are registered via EventSourcingHandler adapters (List) errorTypes, // TODO: wire aggregateServicesSupplier from AkcesAgenticAggregateController (Phase 3) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java index 718f90cf..84768c3e 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java @@ -17,6 +17,9 @@ package org.elasticsoftware.akces.agentic.runtime; +import com.embabel.agent.api.common.autonomy.AgentProcessExecution; +import com.embabel.agent.api.common.autonomy.Autonomy; +import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -34,6 +37,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.function.Supplier; import java.util.stream.Stream; @@ -46,11 +50,16 @@ *

    *
  1. Assembles a bindings {@link Map} containing the command, current aggregate state, * memories, aggregate service records, and condition flags.
    2. - *
  2. Creates an {@link AgentProcess} via - * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} using the - * provided {@link Agent} and the assembled bindings.
    3. - *
  3. Calls {@link AgentProcess#tick()} in a loop until the process reaches an end - * state (completed, failed, terminated, or killed).
    4. + *
  4. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching + * the aggregate name against deployed agent names (exact match or + * {@code {aggregateName}Agent} suffix match).
    5. + *
  5. If a matching agent is found, creates an {@link AgentProcess} via + * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls + * {@link AgentProcess#tick()} in a loop until the process reaches an end state + * (completed, failed, terminated, or killed).
    6. + *
  6. If no matching agent is found, falls back to + * {@link Autonomy#chooseAndAccomplishGoal} to let the Embabel platform decide + * on the best agent and goal combination.
    7. *
  7. Collects {@link DomainEvent} objects placed on the agent's blackboard via * {@link AgentProcessResultTranslator#collectEvents} and returns them as a * {@link Stream}.
    8. @@ -75,9 +84,9 @@ public class AgenticCommandHandlerFunctionAdapter aggregate; + private final String aggregateName; private final CommandType commandType; private final AgentPlatform agentPlatform; - private final Agent agent; private final List> producedDomainEventTypes; private final List> errorEventTypes; private final Supplier> aggregateServicesSupplier; @@ -86,10 +95,10 @@ public class AgenticCommandHandlerFunctionAdapter aggregate, + String aggregateName, CommandType commandType, AgentPlatform agentPlatform, - Agent agent, List> producedDomainEventTypes, List> errorEventTypes, Supplier> aggregateServicesSupplier) { this.aggregate = aggregate; + this.aggregateName = aggregateName; this.commandType = commandType; this.agentPlatform = agentPlatform; - this.agent = agent; this.producedDomainEventTypes = List.copyOf(producedDomainEventTypes); this.errorEventTypes = List.copyOf(errorEventTypes); this.aggregateServicesSupplier = aggregateServicesSupplier; @@ -136,7 +145,7 @@ public AgenticCommandHandlerFunctionAdapter( @SuppressWarnings("unchecked") public Stream apply(@Nonnull C command, S state) { logger.debug("Processing agent-handled command {} for aggregate {}", - commandType.typeName(), aggregate.getClass().getSimpleName()); + commandType.typeName(), aggregateName); Map bindings = new LinkedHashMap<>(); bindings.put("command", command); @@ -151,41 +160,34 @@ public Stream apply(@Nonnull C command, S state) { bindings.put("isExternalEvent", false); bindings.put("hasMemories", !memories.isEmpty()); - AgentProcess agentProcess = - agentPlatform.createAgentProcess(agent, ProcessOptions.DEFAULT, bindings); + Optional resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); + AgentProcess agentProcess; - // Tick to completion with defensive limits so a stuck agent process cannot - // block command handling indefinitely. - final long maxTicks = 10_000L; - final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); - final long deadlineNanos = System.nanoTime() + timeoutNanos; - long tickCount = 0L; - - while (!agentProcess.getFinished() - && tickCount < maxTicks - && System.nanoTime() < deadlineNanos) { - agentProcess.tick(); - tickCount++; - } - - if (!agentProcess.getFinished()) { - logger.error( - "Agent process did not finish within safety limits for command {} on aggregate {}. " + - "aggregateId={}, tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", - commandType.typeName(), - aggregate.getClass().getSimpleName(), - state.getAggregateId(), - tickCount, - maxTicks, - java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), - agentProcess.getStatus()); - throw new IllegalStateException( - "Agent process exceeded execution limits for command " + commandType.typeName() - + " on aggregate " + aggregate.getClass().getSimpleName()); + if (resolvedAgent.isPresent()) { + logger.debug("Resolved agent '{}' for aggregate '{}'", + resolvedAgent.get().getName(), aggregateName); + agentProcess = agentPlatform.createAgentProcess( + resolvedAgent.get(), ProcessOptions.DEFAULT, bindings); + tickToCompletion(agentProcess); + } else { + logger.info("No agent found matching aggregate name '{}'; falling back to Autonomy", + aggregateName); + Autonomy autonomy = agentPlatform.getPlatformServices().autonomy(); + try { + AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( + GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); + agentProcess = execution.getAgentProcess(); + } catch (Exception e) { + logger.error("Autonomy fallback failed for command {} on aggregate {}", + commandType.typeName(), aggregateName, e); + throw new IllegalStateException( + "Autonomy fallback failed for command " + commandType.typeName() + + " on aggregate " + aggregateName, e); + } } logger.debug("Agent process completed with status {} for command {} on aggregate {}", - agentProcess.getStatus(), commandType.typeName(), aggregate.getClass().getSimpleName()); + agentProcess.getStatus(), commandType.typeName(), aggregateName); return (Stream) AgentProcessResultTranslator .collectEvents(agentProcess.getBlackboard(), getAllRegisteredEventTypes()) @@ -237,4 +239,67 @@ private List> getAllRegisteredEventTypes() { all.addAll(errorEventTypes); return all; } + + /** + * Resolves an {@link Agent} from the {@link AgentPlatform} by matching deployed agent + * names against the given aggregate name. + * + *

    Matching rules (checked in order): + *

      + *
    1. Exact match: {@code agent.getName().equals(aggregateName)}
      2. + *
    2. Agent-suffix match: {@code agent.getName().equals(aggregateName + "Agent")}
      3. + *
    + * + * @param platform the agent platform containing deployed agents + * @param aggregateName the aggregate name to match against + * @return an {@link Optional} containing the matched agent, or empty if none found + */ + static Optional resolveAgentByName(AgentPlatform platform, String aggregateName) { + String suffixName = aggregateName + "Agent"; + Optional exact = platform.agents().stream() + .filter(a -> a.getName().equals(aggregateName)) + .findFirst(); + if (exact.isPresent()) { + return exact; + } + return platform.agents().stream() + .filter(a -> a.getName().equals(suffixName)) + .findFirst(); + } + + /** + * Ticks an {@link AgentProcess} to completion with defensive limits so a stuck agent + * process cannot block command or event handling indefinitely. + * + * @param agentProcess the agent process to drive to completion + * @throws IllegalStateException if the process does not finish within the safety limits + */ + private void tickToCompletion(AgentProcess agentProcess) { + final long maxTicks = 10_000L; + final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); + final long deadlineNanos = System.nanoTime() + timeoutNanos; + long tickCount = 0L; + + while (!agentProcess.getFinished() + && tickCount < maxTicks + && System.nanoTime() < deadlineNanos) { + agentProcess.tick(); + tickCount++; + } + + if (!agentProcess.getFinished()) { + logger.error( + "Agent process did not finish within safety limits for command {} on aggregate {}. " + + "tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", + commandType.typeName(), + aggregateName, + tickCount, + maxTicks, + java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), + agentProcess.getStatus()); + throw new IllegalStateException( + "Agent process exceeded execution limits for command " + commandType.typeName() + + " on aggregate " + aggregateName); + } + } } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index 5a07ec43..afafbba1 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -17,6 +17,9 @@ package org.elasticsoftware.akces.agentic.runtime; +import com.embabel.agent.api.common.autonomy.AgentProcessExecution; +import com.embabel.agent.api.common.autonomy.Autonomy; +import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -33,6 +36,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.function.Supplier; import java.util.stream.Stream; @@ -45,11 +49,16 @@ *
      *
    1. Assembles a bindings {@link Map} containing the event, current aggregate state, * memories, aggregate service records, and condition flags.
      2. - *
    2. Creates an {@link AgentProcess} via - * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} using the - * provided {@link Agent} and the assembled bindings.
      3. - *
    3. Calls {@link AgentProcess#tick()} in a loop until the process reaches an end - * state (completed, failed, terminated, or killed).
      4. + *
    4. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching + * the aggregate name against deployed agent names (exact match or + * {@code {aggregateName}Agent} suffix match).
      5. + *
    5. If a matching agent is found, creates an {@link AgentProcess} via + * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls + * {@link AgentProcess#tick()} in a loop until the process reaches an end state + * (completed, failed, terminated, or killed).
      6. + *
    6. If no matching agent is found, falls back to + * {@link Autonomy#chooseAndAccomplishGoal} to let the Embabel platform decide + * on the best agent and goal combination.
      7. *
    7. Collects {@link DomainEvent} objects placed on the agent's blackboard via * {@link AgentProcessResultTranslator#collectEvents} and returns them as a * {@link Stream}.
      8. @@ -74,9 +83,9 @@ public class AgenticEventHandlerFunctionAdapter aggregate; + private final String aggregateName; private final DomainEventType eventType; private final AgentPlatform agentPlatform; - private final Agent agent; private final List> producedDomainEventTypes; private final List> errorEventTypes; private final Supplier> aggregateServicesSupplier; @@ -85,10 +94,10 @@ public class AgenticEventHandlerFunctionAdapter aggregate, + String aggregateName, DomainEventType eventType, AgentPlatform agentPlatform, - Agent agent, List> producedDomainEventTypes, List> errorEventTypes, Supplier> aggregateServicesSupplier) { this.aggregate = aggregate; + this.aggregateName = aggregateName; this.eventType = eventType; this.agentPlatform = agentPlatform; - this.agent = agent; this.producedDomainEventTypes = List.copyOf(producedDomainEventTypes); this.errorEventTypes = List.copyOf(errorEventTypes); this.aggregateServicesSupplier = aggregateServicesSupplier; @@ -135,7 +144,7 @@ public AgenticEventHandlerFunctionAdapter( @SuppressWarnings("unchecked") public Stream apply(@Nonnull InputEvent event, S state) { logger.debug("Processing agent-handled external event {} for aggregate {}", - eventType.typeName(), aggregate.getClass().getSimpleName()); + eventType.typeName(), aggregateName); Map bindings = new LinkedHashMap<>(); bindings.put("event", event); @@ -150,40 +159,35 @@ public Stream apply(@Nonnull InputEvent event, S state) { bindings.put("isExternalEvent", true); bindings.put("hasMemories", !memories.isEmpty()); - AgentProcess agentProcess = - agentPlatform.createAgentProcess(agent, ProcessOptions.DEFAULT, bindings); + Optional resolvedAgent = + AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, aggregateName); + AgentProcess agentProcess; - // Tick to completion with defensive limits so a stuck agent process cannot - // block external event handling indefinitely. - final long maxTicks = 10_000L; - final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); - final long deadlineNanos = System.nanoTime() + timeoutNanos; - long tickCount = 0L; - - while (!agentProcess.getFinished() - && tickCount < maxTicks - && System.nanoTime() < deadlineNanos) { - agentProcess.tick(); - tickCount++; + if (resolvedAgent.isPresent()) { + logger.debug("Resolved agent '{}' for aggregate '{}'", + resolvedAgent.get().getName(), aggregateName); + agentProcess = agentPlatform.createAgentProcess( + resolvedAgent.get(), ProcessOptions.DEFAULT, bindings); + tickToCompletion(agentProcess); + } else { + logger.info("No agent found matching aggregate name '{}'; falling back to Autonomy", + aggregateName); + Autonomy autonomy = agentPlatform.getPlatformServices().autonomy(); + try { + AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( + GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); + agentProcess = execution.getAgentProcess(); + } catch (Exception e) { + logger.error("Autonomy fallback failed for event {} on aggregate {}", + eventType.typeName(), aggregateName, e); + throw new IllegalStateException( + "Autonomy fallback failed for event " + eventType.typeName() + + " on aggregate " + aggregateName, e); + } } - if (!agentProcess.getFinished()) { - logger.error( - "Agent process did not finish within safety limits for event {} on aggregate {}. " + - "aggregateId={}, tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", - eventType.typeName(), - aggregate.getClass().getSimpleName(), - state.getAggregateId(), - tickCount, - maxTicks, - java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), - agentProcess.getStatus()); - throw new IllegalStateException( - "Agent process exceeded execution limits for event " + eventType.typeName() - + " on aggregate " + aggregate.getClass().getSimpleName()); - } logger.debug("Agent process completed with status {} for event {} on aggregate {}", - agentProcess.getStatus(), eventType.typeName(), aggregate.getClass().getSimpleName()); + agentProcess.getStatus(), eventType.typeName(), aggregateName); return (Stream) AgentProcessResultTranslator .collectEvents(agentProcess.getBlackboard(), getAllRegisteredEventTypes()) @@ -234,4 +238,40 @@ private List> getAllRegisteredEventTypes() { all.addAll(errorEventTypes); return all; } + + /** + * Ticks an {@link AgentProcess} to completion with defensive limits so a stuck agent + * process cannot block event handling indefinitely. + * + * @param agentProcess the agent process to drive to completion + * @throws IllegalStateException if the process does not finish within the safety limits + */ + private void tickToCompletion(AgentProcess agentProcess) { + final long maxTicks = 10_000L; + final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); + final long deadlineNanos = System.nanoTime() + timeoutNanos; + long tickCount = 0L; + + while (!agentProcess.getFinished() + && tickCount < maxTicks + && System.nanoTime() < deadlineNanos) { + agentProcess.tick(); + tickCount++; + } + + if (!agentProcess.getFinished()) { + logger.error( + "Agent process did not finish within safety limits for event {} on aggregate {}. " + + "tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", + eventType.typeName(), + aggregateName, + tickCount, + maxTicks, + java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), + agentProcess.getStatus()); + throw new IllegalStateException( + "Agent process exceeded execution limits for event " + eventType.typeName() + + " on aggregate " + aggregateName); + } + } } From 74c73226cc3b8c14003bc8b7e7651f461e16a0c9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 10:50:06 +0000 Subject: [PATCH 03/10] Add unit tests for agent resolution and Autonomy fallback Add AgenticCommandHandlerFunctionAdapterTest (12 tests) and AgenticEventHandlerFunctionAdapterTest (7 tests) covering: - Exact name match resolution - Suffix name match resolution - Autonomy chooseAndAccomplishGoal() fallback - Exception wrapping for Autonomy failures - Exact match precedence over suffix match - resolveAgentByName static helper - Basic adapter properties Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/e6ed61c0-c5ac-4bbc-9cab-5e7276244968 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- ...nticCommandHandlerFunctionAdapterTest.java | 306 ++++++++++++++++++ ...genticEventHandlerFunctionAdapterTest.java | 231 +++++++++++++ 2 files changed, 537 insertions(+) create mode 100644 main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java create mode 100644 main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java new file mode 100644 index 00000000..1e9f8823 --- /dev/null +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java @@ -0,0 +1,306 @@ +/* + * 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.api.common.PlatformServices; +import com.embabel.agent.api.common.autonomy.AgentProcessExecution; +import com.embabel.agent.api.common.autonomy.Autonomy; +import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; +import com.embabel.agent.api.common.autonomy.ProcessExecutionException; +import com.embabel.agent.core.Agent; +import com.embabel.agent.core.AgentPlatform; +import com.embabel.agent.core.AgentProcess; +import com.embabel.agent.core.Blackboard; +import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.aggregate.*; +import org.elasticsoftware.akces.annotations.CommandInfo; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.commands.Command; +import org.elasticsoftware.akces.events.DomainEvent; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +/** + * Unit tests for {@link AgenticCommandHandlerFunctionAdapter}, verifying the runtime + * agent resolution by aggregate name and the {@link Autonomy} fallback when no matching + * agent is found. + */ +@ExtendWith(MockitoExtension.class) +class AgenticCommandHandlerFunctionAdapterTest { + + @CommandInfo(type = "TestCommand", version = 1) + record TestCommand(String id) implements Command { + @Override + public String getAggregateId() { + return id; + } + } + + @DomainEventInfo(type = "TestEvent", version = 1) + record TestEvent(String id) implements DomainEvent { + @Override + public String getAggregateId() { + return id; + } + } + + record TestState(String id) implements AggregateState { + @Override + public String getAggregateId() { + return id; + } + } + + @Mock + private AgenticAggregate aggregate; + + @Mock + private AgentPlatform agentPlatform; + + @Mock + private AgentProcess agentProcess; + + @Mock + private Blackboard blackboard; + + @Mock + private PlatformServices platformServices; + + @Mock + private Autonomy autonomy; + + private final CommandType commandType = + new CommandType<>("TestCommand", 1, TestCommand.class, false, false, false); + + @BeforeEach + void setUp() { + lenient().when(agentProcess.getBlackboard()).thenReturn(blackboard); + lenient().when(blackboard.get(any(String.class))).thenReturn(null); + } + + // ------------------------------------------------------------------------- + // Agent resolution by exact name + // ------------------------------------------------------------------------- + + @Test + void shouldResolveAgentByExactNameMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("TestAggregate"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + when(agentPlatform.createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "TestAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply(new TestCommand("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(agentPlatform).createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class)); + } + + // ------------------------------------------------------------------------- + // Agent resolution by suffix name + // ------------------------------------------------------------------------- + + @Test + void shouldResolveAgentBySuffixNameMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("TestAggregateAgent"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + when(agentPlatform.createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "TestAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply(new TestCommand("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(agentPlatform).createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class)); + } + + // ------------------------------------------------------------------------- + // Autonomy fallback + // ------------------------------------------------------------------------- + + @Test + void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException { + when(agentPlatform.agents()).thenReturn(List.of()); + when(agentPlatform.getPlatformServices()).thenReturn(platformServices); + when(platformServices.autonomy()).thenReturn(autonomy); + + AgentProcessExecution execution = mock(AgentProcessExecution.class); + when(execution.getAgentProcess()).thenReturn(agentProcess); + when(autonomy.chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) + .thenReturn(execution); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply(new TestCommand("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(autonomy).chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); + verify(agentPlatform, never()).createAgentProcess(any(), any(), any(Map.class)); + } + + @Test + void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutionException { + when(agentPlatform.agents()).thenReturn(List.of()); + when(agentPlatform.getPlatformServices()).thenReturn(platformServices); + when(platformServices.autonomy()).thenReturn(autonomy); + when(autonomy.chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) + .thenThrow(new RuntimeException("Autonomy failed")); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThatThrownBy(() -> adapter.apply(new TestCommand("agg-1"), new TestState("agg-1"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Autonomy fallback failed") + .hasMessageContaining("TestCommand") + .hasMessageContaining("UnknownAggregate"); + } + + // ------------------------------------------------------------------------- + // Exact match takes precedence over suffix match + // ------------------------------------------------------------------------- + + @Test + void shouldPreferExactMatchOverSuffixMatch() { + Agent exactAgent = mock(Agent.class); + when(exactAgent.getName()).thenReturn("MyAggregate"); + + Agent suffixAgent = mock(Agent.class); + lenient().when(suffixAgent.getName()).thenReturn("MyAggregateAgent"); + + when(agentPlatform.agents()).thenReturn(List.of(suffixAgent, exactAgent)); + when(agentPlatform.createAgentProcess(eq(exactAgent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "MyAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + adapter.apply(new TestCommand("agg-1"), new TestState("agg-1")); + + verify(agentPlatform).createAgentProcess(eq(exactAgent), eq(ProcessOptions.DEFAULT), any(Map.class)); + } + + // ------------------------------------------------------------------------- + // resolveAgentByName static helper + // ------------------------------------------------------------------------- + + @Test + void resolveAgentByNameShouldReturnEmptyWhenNoAgentsDeployed() { + when(agentPlatform.agents()).thenReturn(List.of()); + + assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) + .isEmpty(); + } + + @Test + void resolveAgentByNameShouldReturnEmptyWhenNoNameMatches() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("SomeOtherAgent"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + + assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) + .isEmpty(); + } + + @Test + void resolveAgentByNameShouldFindExactMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("Wallet"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + + assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Wallet")) + .contains(agent); + } + + @Test + void resolveAgentByNameShouldFindSuffixMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("WalletAgent"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + + assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Wallet")) + .contains(agent); + } + + // ------------------------------------------------------------------------- + // Basic adapter properties + // ------------------------------------------------------------------------- + + @Test + void isCreateShouldAlwaysReturnFalse() { + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "Test", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.isCreate()).isFalse(); + } + + @Test + void getCommandTypeShouldReturnConfiguredType() { + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "Test", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.getCommandType()).isSameAs(commandType); + } + + @Test + void getAggregateShouldReturnConfiguredAggregate() { + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "Test", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.getAggregate()).isSameAs(aggregate); + } +} diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java new file mode 100644 index 00000000..759c4a16 --- /dev/null +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java @@ -0,0 +1,231 @@ +/* + * 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.api.common.PlatformServices; +import com.embabel.agent.api.common.autonomy.AgentProcessExecution; +import com.embabel.agent.api.common.autonomy.Autonomy; +import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; +import com.embabel.agent.api.common.autonomy.ProcessExecutionException; +import com.embabel.agent.core.Agent; +import com.embabel.agent.core.AgentPlatform; +import com.embabel.agent.core.AgentProcess; +import com.embabel.agent.core.Blackboard; +import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.aggregate.*; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.events.DomainEvent; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +/** + * Unit tests for {@link AgenticEventHandlerFunctionAdapter}, verifying the runtime + * agent resolution by aggregate name and the {@link Autonomy} fallback when no matching + * agent is found. + */ +@ExtendWith(MockitoExtension.class) +class AgenticEventHandlerFunctionAdapterTest { + + @DomainEventInfo(type = "ExternalEvent", version = 1) + record TestExternalEvent(String id) implements DomainEvent { + @Override + public String getAggregateId() { + return id; + } + } + + record TestState(String id) implements AggregateState { + @Override + public String getAggregateId() { + return id; + } + } + + @Mock + private AgenticAggregate aggregate; + + @Mock + private AgentPlatform agentPlatform; + + @Mock + private AgentProcess agentProcess; + + @Mock + private Blackboard blackboard; + + @Mock + private PlatformServices platformServices; + + @Mock + private Autonomy autonomy; + + private final DomainEventType eventType = + new DomainEventType<>("ExternalEvent", 1, TestExternalEvent.class, false, true, false, false); + + @BeforeEach + void setUp() { + lenient().when(agentProcess.getBlackboard()).thenReturn(blackboard); + lenient().when(blackboard.get(any(String.class))).thenReturn(null); + } + + // ------------------------------------------------------------------------- + // Agent resolution by exact name + // ------------------------------------------------------------------------- + + @Test + void shouldResolveAgentByExactNameMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("TestAggregate"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + when(agentPlatform.createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "TestAggregate", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply( + new TestExternalEvent("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(agentPlatform).createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class)); + } + + // ------------------------------------------------------------------------- + // Agent resolution by suffix name + // ------------------------------------------------------------------------- + + @Test + void shouldResolveAgentBySuffixNameMatch() { + Agent agent = mock(Agent.class); + when(agent.getName()).thenReturn("TestAggregateAgent"); + when(agentPlatform.agents()).thenReturn(List.of(agent)); + when(agentPlatform.createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "TestAggregate", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply( + new TestExternalEvent("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(agentPlatform).createAgentProcess(eq(agent), eq(ProcessOptions.DEFAULT), any(Map.class)); + } + + // ------------------------------------------------------------------------- + // Autonomy fallback + // ------------------------------------------------------------------------- + + @Test + void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException { + when(agentPlatform.agents()).thenReturn(List.of()); + when(agentPlatform.getPlatformServices()).thenReturn(platformServices); + when(platformServices.autonomy()).thenReturn(autonomy); + + AgentProcessExecution execution = mock(AgentProcessExecution.class); + when(execution.getAgentProcess()).thenReturn(agentProcess); + when(autonomy.chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) + .thenReturn(execution); + when(agentProcess.getStatus()).thenReturn(null); + + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + Stream result = adapter.apply( + new TestExternalEvent("agg-1"), new TestState("agg-1")); + + assertThat(result).isEmpty(); + verify(autonomy).chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); + verify(agentPlatform, never()).createAgentProcess(any(), any(), any(Map.class)); + } + + @Test + void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutionException { + when(agentPlatform.agents()).thenReturn(List.of()); + when(agentPlatform.getPlatformServices()).thenReturn(platformServices); + when(platformServices.autonomy()).thenReturn(autonomy); + when(autonomy.chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) + .thenThrow(new RuntimeException("Autonomy failed")); + + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThatThrownBy(() -> adapter.apply( + new TestExternalEvent("agg-1"), new TestState("agg-1"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Autonomy fallback failed") + .hasMessageContaining("ExternalEvent") + .hasMessageContaining("UnknownAggregate"); + } + + // ------------------------------------------------------------------------- + // Basic adapter properties + // ------------------------------------------------------------------------- + + @Test + void isCreateShouldAlwaysReturnFalse() { + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "Test", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.isCreate()).isFalse(); + } + + @Test + void getEventTypeShouldReturnConfiguredType() { + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "Test", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.getEventType()).isSameAs(eventType); + } + + @Test + void getAggregateShouldReturnConfiguredAggregate() { + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "Test", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThat(adapter.getAggregate()).isSameAs(aggregate); + } +} From d17b078692c24a7d07b60d38bf372c988240a179 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 10:56:46 +0000 Subject: [PATCH 04/10] Address code review: catch ProcessExecutionException specifically, fix test mocks - Narrow exception catch from generic Exception to ProcessExecutionException - Use doThrow() for Kotlin final method stubbing in tests - Provide non-null AgentProcess to ProcessExecutionFailedException constructor Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/e6ed61c0-c5ac-4bbc-9cab-5e7276244968 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../runtime/AgenticCommandHandlerFunctionAdapter.java | 3 ++- .../runtime/AgenticEventHandlerFunctionAdapter.java | 3 ++- .../AgenticCommandHandlerFunctionAdapterTest.java | 9 ++++++--- .../runtime/AgenticEventHandlerFunctionAdapterTest.java | 9 ++++++--- 4 files changed, 16 insertions(+), 8 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java index 84768c3e..1d4df78c 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java @@ -20,6 +20,7 @@ import com.embabel.agent.api.common.autonomy.AgentProcessExecution; import com.embabel.agent.api.common.autonomy.Autonomy; import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; +import com.embabel.agent.api.common.autonomy.ProcessExecutionException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -177,7 +178,7 @@ public Stream apply(@Nonnull C command, S state) { AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); agentProcess = execution.getAgentProcess(); - } catch (Exception e) { + } catch (ProcessExecutionException e) { logger.error("Autonomy fallback failed for command {} on aggregate {}", commandType.typeName(), aggregateName, e); throw new IllegalStateException( diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index afafbba1..df57ac96 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -20,6 +20,7 @@ import com.embabel.agent.api.common.autonomy.AgentProcessExecution; import com.embabel.agent.api.common.autonomy.Autonomy; import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; +import com.embabel.agent.api.common.autonomy.ProcessExecutionException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -177,7 +178,7 @@ public Stream apply(@Nonnull InputEvent event, S state) { AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); agentProcess = execution.getAgentProcess(); - } catch (Exception e) { + } catch (ProcessExecutionException e) { logger.error("Autonomy fallback failed for event {} on aggregate {}", eventType.typeName(), aggregateName, e); throw new IllegalStateException( diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java index 1e9f8823..dd9fe5f1 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java @@ -22,6 +22,7 @@ import com.embabel.agent.api.common.autonomy.Autonomy; import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; import com.embabel.agent.api.common.autonomy.ProcessExecutionException; +import com.embabel.agent.api.common.autonomy.ProcessExecutionFailedException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -189,9 +190,11 @@ void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutio when(agentPlatform.agents()).thenReturn(List.of()); when(agentPlatform.getPlatformServices()).thenReturn(platformServices); when(platformServices.autonomy()).thenReturn(autonomy); - when(autonomy.chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) - .thenThrow(new RuntimeException("Autonomy failed")); + + AgentProcess failedProcess = mock(AgentProcess.class); + doThrow(new ProcessExecutionFailedException(failedProcess, "Autonomy failed")) + .when(autonomy).chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); var adapter = new AgenticCommandHandlerFunctionAdapter<>( aggregate, "UnknownAggregate", commandType, agentPlatform, diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java index 759c4a16..96dd0910 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java @@ -22,6 +22,7 @@ import com.embabel.agent.api.common.autonomy.Autonomy; import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; import com.embabel.agent.api.common.autonomy.ProcessExecutionException; +import com.embabel.agent.api.common.autonomy.ProcessExecutionFailedException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -182,9 +183,11 @@ void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutio when(agentPlatform.agents()).thenReturn(List.of()); when(agentPlatform.getPlatformServices()).thenReturn(platformServices); when(platformServices.autonomy()).thenReturn(autonomy); - when(autonomy.chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) - .thenThrow(new RuntimeException("Autonomy failed")); + + AgentProcess failedProcess = mock(AgentProcess.class); + doThrow(new ProcessExecutionFailedException(failedProcess, "Autonomy failed")) + .when(autonomy).chooseAndAccomplishGoal( + any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); var adapter = new AgenticEventHandlerFunctionAdapter<>( aggregate, "UnknownAggregate", eventType, agentPlatform, From 5316887e672ca10274b9ccdee2686d710465c00b Mon Sep 17 00:00:00 2001 From: Joost van de Wijgerd Date: Fri, 10 Apr 2026 14:10:27 +0200 Subject: [PATCH 05/10] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../AgenticCommandHandlerFunctionAdapter.java | 45 ++++++++++++----- .../AgenticEventHandlerFunctionAdapter.java | 49 ++++++++++++++----- 2 files changed, 72 insertions(+), 22 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java index 1d4df78c..978efb55 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java @@ -257,25 +257,46 @@ private List> getAllRegisteredEventTypes() { */ static Optional resolveAgentByName(AgentPlatform platform, String aggregateName) { String suffixName = aggregateName + "Agent"; - Optional exact = platform.agents().stream() - .filter(a -> a.getName().equals(aggregateName)) - .findFirst(); - if (exact.isPresent()) { - return exact; + Collection agents = platform.agents(); + Agent suffixMatch = null; + + for (Agent agent : agents) { + String agentName = agent.getName(); + if (agentName.equals(aggregateName)) { + return Optional.of(agent); + } + if (suffixMatch == null && agentName.equals(suffixName)) { + suffixMatch = agent; + } } - return platform.agents().stream() - .filter(a -> a.getName().equals(suffixName)) - .findFirst(); + + return Optional.ofNullable(suffixMatch); } /** * Ticks an {@link AgentProcess} to completion with defensive limits so a stuck agent * process cannot block command or event handling indefinitely. * + *

      This overload preserves the existing call sites and attempts to resolve the + * aggregate identifier from SLF4J MDC using the {@code aggregateId} key. + * * @param agentProcess the agent process to drive to completion * @throws IllegalStateException if the process does not finish within the safety limits */ private void tickToCompletion(AgentProcess agentProcess) { + String aggregateId = org.slf4j.MDC.get("aggregateId"); + tickToCompletion(agentProcess, aggregateId != null ? aggregateId : ""); + } + + /** + * Ticks an {@link AgentProcess} to completion with defensive limits so a stuck agent + * process cannot block command or event handling indefinitely. + * + * @param agentProcess the agent process to drive to completion + * @param aggregateId the aggregate instance identifier for diagnostic logging + * @throws IllegalStateException if the process does not finish within the safety limits + */ + private void tickToCompletion(AgentProcess agentProcess, String aggregateId) { final long maxTicks = 10_000L; final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); final long deadlineNanos = System.nanoTime() + timeoutNanos; @@ -290,17 +311,19 @@ private void tickToCompletion(AgentProcess agentProcess) { if (!agentProcess.getFinished()) { logger.error( - "Agent process did not finish within safety limits for command {} on aggregate {}. " + - "tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", + "Agent process did not finish within safety limits for command {} on aggregate {} " + + "with aggregateId {}. tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", commandType.typeName(), aggregateName, + aggregateId, tickCount, maxTicks, java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), agentProcess.getStatus()); throw new IllegalStateException( "Agent process exceeded execution limits for command " + commandType.typeName() - + " on aggregate " + aggregateName); + + " on aggregate " + aggregateName + + " with aggregateId " + aggregateId); } } } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index df57ac96..9758c212 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -141,6 +141,12 @@ public AgenticEventHandlerFunctionAdapter( * @return a stream of domain events produced by the agent; may be empty */ @Nonnull + private static Optional resolveAgentByName(AgentPlatform agentPlatform, String aggregateName) { + return agentPlatform.getAgents().stream() + .filter(agent -> aggregateName.equals(agent.getName())) + .findFirst(); + } + @Override @SuppressWarnings("unchecked") public Stream apply(@Nonnull InputEvent event, S state) { @@ -160,8 +166,7 @@ public Stream apply(@Nonnull InputEvent event, S state) { bindings.put("isExternalEvent", true); bindings.put("hasMemories", !memories.isEmpty()); - Optional resolvedAgent = - AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, aggregateName); + Optional resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); AgentProcess agentProcess; if (resolvedAgent.isPresent()) { @@ -248,13 +253,34 @@ private List> getAllRegisteredEventTypes() { * @throws IllegalStateException if the process does not finish within the safety limits */ private void tickToCompletion(AgentProcess agentProcess) { - final long maxTicks = 10_000L; - final long timeoutNanos = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); - final long deadlineNanos = System.nanoTime() + timeoutNanos; + AgenticProcessTickHelper.tickToCompletion( + agentProcess, + logger, + "event", + eventType.typeName(), + aggregateName); + } +} + +final class AgenticProcessTickHelper { + + private static final long MAX_TICKS = 10_000L; + private static final long TIMEOUT_NANOS = java.util.concurrent.TimeUnit.SECONDS.toNanos(30); + + private AgenticProcessTickHelper() { + } + + static void tickToCompletion( + AgentProcess agentProcess, + Logger logger, + String handledTypeLabel, + String handledTypeName, + String aggregateName) { + final long deadlineNanos = System.nanoTime() + TIMEOUT_NANOS; long tickCount = 0L; while (!agentProcess.getFinished() - && tickCount < maxTicks + && tickCount < MAX_TICKS && System.nanoTime() < deadlineNanos) { agentProcess.tick(); tickCount++; @@ -262,16 +288,17 @@ private void tickToCompletion(AgentProcess agentProcess) { if (!agentProcess.getFinished()) { logger.error( - "Agent process did not finish within safety limits for event {} on aggregate {}. " + + "Agent process did not finish within safety limits for {} {} on aggregate {}. " + "tickCount={}, maxTicks={}, timeoutSeconds={}, status={}", - eventType.typeName(), + handledTypeLabel, + handledTypeName, aggregateName, tickCount, - maxTicks, - java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(timeoutNanos), + MAX_TICKS, + java.util.concurrent.TimeUnit.NANOSECONDS.toSeconds(TIMEOUT_NANOS), agentProcess.getStatus()); throw new IllegalStateException( - "Agent process exceeded execution limits for event " + eventType.typeName() + "Agent process exceeded execution limits for " + handledTypeLabel + " " + handledTypeName + " on aggregate " + aggregateName); } } From f1d995e9c8d992e0060898269ecbbbafa939628d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 12:17:33 +0000 Subject: [PATCH 06/10] Fix build: use agents() instead of getAgents() in AgenticEventHandlerFunctionAdapter The code review suggestion commit introduced getAgents() which does not exist on the AgentPlatform interface. The correct method is agents(). Also restored single-pass agent resolution logic with suffix match support that was lost in the code review suggestion. Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/e6bbbe64-67fe-4997-9046-7e96a9a43702 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../AgenticEventHandlerFunctionAdapter.java | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index 9758c212..b3cf9299 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -142,9 +142,21 @@ public AgenticEventHandlerFunctionAdapter( */ @Nonnull private static Optional resolveAgentByName(AgentPlatform agentPlatform, String aggregateName) { - return agentPlatform.getAgents().stream() - .filter(agent -> aggregateName.equals(agent.getName())) - .findFirst(); + String suffixName = aggregateName + "Agent"; + Collection agents = agentPlatform.agents(); + Agent suffixMatch = null; + + for (Agent agent : agents) { + String agentName = agent.getName(); + if (agentName.equals(aggregateName)) { + return Optional.of(agent); + } + if (suffixMatch == null && agentName.equals(suffixName)) { + suffixMatch = agent; + } + } + + return Optional.ofNullable(suffixMatch); } @Override From 15959761b1bda30fbc57bfba88ce52832743952e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 12:43:12 +0000 Subject: [PATCH 07/10] Fix javadoc: remove stale {@link Agent} references in AgenticAggregateRuntimeFactory The class-level javadoc still referenced {@link Agent} which was no longer imported after agent resolution moved to the adapters. Updated the paragraph to describe the current lazy-resolution behavior. Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/d46ab258-b06a-4d2a-b1ed-ccc2c1db2aae Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../agentic/beans/AgenticAggregateRuntimeFactory.java | 7 +++---- 1 file changed, 3 insertions(+), 4 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 34ba40d1..c087e3d6 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 @@ -73,10 +73,9 @@ * Embabel agent platform.

      * *

      When {@code agentHandledCommands} or {@code agentHandledEvents} are declared, the - * factory looks up an {@link Agent} bean named {@code {aggregateName}Agent} from the - * {@link ApplicationContext}. This {@link Agent} is provided by the implementing - * application and must be registered as a Spring bean before this factory is invoked. - * A fatal error is raised if the bean cannot be found.

      + * factory creates adapter instances that resolve the matching agent at runtime from + * the {@link AgentPlatform} by aggregate name. No eager bean lookup is performed at + * startup — agent resolution is deferred to command/event processing time.

      * *

      Produces a {@link KafkaAgenticAggregateRuntime} wrapping the internally built * {@link KafkaAggregateRuntime}, adding memory-aware and agent-platform operations.

      From 9477ef61ebc1d539b091421fd0a75d978ae53ad8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 13:13:49 +0000 Subject: [PATCH 08/10] Replace Autonomy fallback with default agent resolution The Autonomy.chooseAndAccomplishGoal() approach blocks the thread until completion, causing Kafka consumer timeouts. Replace it with a default agent fallback: when no agent matches the aggregate name (exact or suffix), the first available agent from the platform is used instead. - Remove Autonomy imports and fallback code from both adapters - Change resolveAgentByName to return Agent directly (not Optional) - Throw IllegalStateException only when no agents are deployed at all - Update tests to verify default agent and no-agents-deployed scenarios Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/8cf48950-f35e-46f6-971b-ec135e834c54 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../AgenticCommandHandlerFunctionAdapter.java | 72 +++++++++---------- .../AgenticEventHandlerFunctionAdapter.java | 65 +++++++---------- ...nticCommandHandlerFunctionAdapterTest.java | 65 ++++++----------- ...genticEventHandlerFunctionAdapterTest.java | 49 ++++--------- 4 files changed, 92 insertions(+), 159 deletions(-) diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java index 978efb55..021661f1 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java @@ -17,10 +17,6 @@ package org.elasticsoftware.akces.agentic.runtime; -import com.embabel.agent.api.common.autonomy.AgentProcessExecution; -import com.embabel.agent.api.common.autonomy.Autonomy; -import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; -import com.embabel.agent.api.common.autonomy.ProcessExecutionException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -38,7 +34,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.function.Supplier; import java.util.stream.Stream; @@ -53,14 +48,12 @@ * memories, aggregate service records, and condition flags. *
    8. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching * the aggregate name against deployed agent names (exact match or - * {@code {aggregateName}Agent} suffix match).
      9. - *
    9. If a matching agent is found, creates an {@link AgentProcess} via + * {@code {aggregateName}Agent} suffix match). If no match is found, the first + * available agent from the platform is used as a default.
      10. + *
    10. Creates an {@link AgentProcess} via * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls * {@link AgentProcess#tick()} in a loop until the process reaches an end state * (completed, failed, terminated, or killed).
      11. - *
    11. If no matching agent is found, falls back to - * {@link Autonomy#chooseAndAccomplishGoal} to let the Embabel platform decide - * on the best agent and goal combination.
      12. *
    12. Collects {@link DomainEvent} objects placed on the agent's blackboard via * {@link AgentProcessResultTranslator#collectEvents} and returns them as a * {@link Stream}.
      13. @@ -161,31 +154,12 @@ public Stream apply(@Nonnull C command, S state) { bindings.put("isExternalEvent", false); bindings.put("hasMemories", !memories.isEmpty()); - Optional resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); - AgentProcess agentProcess; - - if (resolvedAgent.isPresent()) { - logger.debug("Resolved agent '{}' for aggregate '{}'", - resolvedAgent.get().getName(), aggregateName); - agentProcess = agentPlatform.createAgentProcess( - resolvedAgent.get(), ProcessOptions.DEFAULT, bindings); - tickToCompletion(agentProcess); - } else { - logger.info("No agent found matching aggregate name '{}'; falling back to Autonomy", - aggregateName); - Autonomy autonomy = agentPlatform.getPlatformServices().autonomy(); - try { - AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( - GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); - agentProcess = execution.getAgentProcess(); - } catch (ProcessExecutionException e) { - logger.error("Autonomy fallback failed for command {} on aggregate {}", - commandType.typeName(), aggregateName, e); - throw new IllegalStateException( - "Autonomy fallback failed for command " + commandType.typeName() - + " on aggregate " + aggregateName, e); - } - } + Agent resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); + logger.debug("Resolved agent '{}' for aggregate '{}'", + resolvedAgent.getName(), aggregateName); + AgentProcess agentProcess = agentPlatform.createAgentProcess( + resolvedAgent, ProcessOptions.DEFAULT, bindings); + tickToCompletion(agentProcess); logger.debug("Agent process completed with status {} for command {} on aggregate {}", agentProcess.getStatus(), commandType.typeName(), aggregateName); @@ -243,19 +217,22 @@ private List> getAllRegisteredEventTypes() { /** * Resolves an {@link Agent} from the {@link AgentPlatform} by matching deployed agent - * names against the given aggregate name. + * names against the given aggregate name. If no match is found, the first available + * agent from the platform is returned as a default. * *

      Matching rules (checked in order): *

        *
      1. Exact match: {@code agent.getName().equals(aggregateName)}
        2. *
      2. Agent-suffix match: {@code agent.getName().equals(aggregateName + "Agent")}
        3. + *
      3. Default: first available agent from the platform
        4. *
      * * @param platform the agent platform containing deployed agents * @param aggregateName the aggregate name to match against - * @return an {@link Optional} containing the matched agent, or empty if none found + * @return the matched or default agent; never {@code null} + * @throws IllegalStateException if no agents are deployed on the platform */ - static Optional resolveAgentByName(AgentPlatform platform, String aggregateName) { + static Agent resolveAgentByName(AgentPlatform platform, String aggregateName) { String suffixName = aggregateName + "Agent"; Collection agents = platform.agents(); Agent suffixMatch = null; @@ -263,14 +240,29 @@ static Optional resolveAgentByName(AgentPlatform platform, String aggrega for (Agent agent : agents) { String agentName = agent.getName(); if (agentName.equals(aggregateName)) { - return Optional.of(agent); + return agent; } if (suffixMatch == null && agentName.equals(suffixName)) { suffixMatch = agent; } } - return Optional.ofNullable(suffixMatch); + if (suffixMatch != null) { + return suffixMatch; + } + + // No name match — use the first available agent as default + if (!agents.isEmpty()) { + Agent defaultAgent = agents.iterator().next(); + logger.info("No agent found matching aggregate name '{}'; using default agent '{}'", + aggregateName, defaultAgent.getName()); + return defaultAgent; + } + + throw new IllegalStateException( + "No agents are deployed on the AgentPlatform. " + + "At least one agent must be available to handle aggregate '" + + aggregateName + "'."); } /** diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index b3cf9299..ba62f465 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -17,10 +17,6 @@ package org.elasticsoftware.akces.agentic.runtime; -import com.embabel.agent.api.common.autonomy.AgentProcessExecution; -import com.embabel.agent.api.common.autonomy.Autonomy; -import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; -import com.embabel.agent.api.common.autonomy.ProcessExecutionException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -37,7 +33,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.function.Supplier; import java.util.stream.Stream; @@ -52,14 +47,12 @@ * memories, aggregate service records, and condition flags. *
    13. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching * the aggregate name against deployed agent names (exact match or - * {@code {aggregateName}Agent} suffix match).
      14. - *
    14. If a matching agent is found, creates an {@link AgentProcess} via + * {@code {aggregateName}Agent} suffix match). If no match is found, the first + * available agent from the platform is used as a default.
      15. + *
    15. Creates an {@link AgentProcess} via * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls * {@link AgentProcess#tick()} in a loop until the process reaches an end state * (completed, failed, terminated, or killed).
      16. - *
    16. If no matching agent is found, falls back to - * {@link Autonomy#chooseAndAccomplishGoal} to let the Embabel platform decide - * on the best agent and goal combination.
      17. *
    17. Collects {@link DomainEvent} objects placed on the agent's blackboard via * {@link AgentProcessResultTranslator#collectEvents} and returns them as a * {@link Stream}.
      18. @@ -141,7 +134,7 @@ public AgenticEventHandlerFunctionAdapter( * @return a stream of domain events produced by the agent; may be empty */ @Nonnull - private static Optional resolveAgentByName(AgentPlatform agentPlatform, String aggregateName) { + private static Agent resolveAgentByName(AgentPlatform agentPlatform, String aggregateName) { String suffixName = aggregateName + "Agent"; Collection agents = agentPlatform.agents(); Agent suffixMatch = null; @@ -149,14 +142,29 @@ private static Optional resolveAgentByName(AgentPlatform agentPlatform, S for (Agent agent : agents) { String agentName = agent.getName(); if (agentName.equals(aggregateName)) { - return Optional.of(agent); + return agent; } if (suffixMatch == null && agentName.equals(suffixName)) { suffixMatch = agent; } } - return Optional.ofNullable(suffixMatch); + if (suffixMatch != null) { + return suffixMatch; + } + + // No name match — use the first available agent as default + if (!agents.isEmpty()) { + Agent defaultAgent = agents.iterator().next(); + logger.info("No agent found matching aggregate name '{}'; using default agent '{}'", + aggregateName, defaultAgent.getName()); + return defaultAgent; + } + + throw new IllegalStateException( + "No agents are deployed on the AgentPlatform. " + + "At least one agent must be available to handle aggregate '" + + aggregateName + "'."); } @Override @@ -178,31 +186,12 @@ public Stream apply(@Nonnull InputEvent event, S state) { bindings.put("isExternalEvent", true); bindings.put("hasMemories", !memories.isEmpty()); - Optional resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); - AgentProcess agentProcess; - - if (resolvedAgent.isPresent()) { - logger.debug("Resolved agent '{}' for aggregate '{}'", - resolvedAgent.get().getName(), aggregateName); - agentProcess = agentPlatform.createAgentProcess( - resolvedAgent.get(), ProcessOptions.DEFAULT, bindings); - tickToCompletion(agentProcess); - } else { - logger.info("No agent found matching aggregate name '{}'; falling back to Autonomy", - aggregateName); - Autonomy autonomy = agentPlatform.getPlatformServices().autonomy(); - try { - AgentProcessExecution execution = autonomy.chooseAndAccomplishGoal( - GoalChoiceApprover.Companion.getAPPROVE_ALL(), agentPlatform, bindings); - agentProcess = execution.getAgentProcess(); - } catch (ProcessExecutionException e) { - logger.error("Autonomy fallback failed for event {} on aggregate {}", - eventType.typeName(), aggregateName, e); - throw new IllegalStateException( - "Autonomy fallback failed for event " + eventType.typeName() - + " on aggregate " + aggregateName, e); - } - } + Agent resolvedAgent = resolveAgentByName(agentPlatform, aggregateName); + logger.debug("Resolved agent '{}' for aggregate '{}'", + resolvedAgent.getName(), aggregateName); + AgentProcess agentProcess = agentPlatform.createAgentProcess( + resolvedAgent, ProcessOptions.DEFAULT, bindings); + tickToCompletion(agentProcess); logger.debug("Agent process completed with status {} for event {} on aggregate {}", agentProcess.getStatus(), eventType.typeName(), aggregateName); diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java index dd9fe5f1..79523ad1 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java @@ -17,12 +17,6 @@ package org.elasticsoftware.akces.agentic.runtime; -import com.embabel.agent.api.common.PlatformServices; -import com.embabel.agent.api.common.autonomy.AgentProcessExecution; -import com.embabel.agent.api.common.autonomy.Autonomy; -import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; -import com.embabel.agent.api.common.autonomy.ProcessExecutionException; -import com.embabel.agent.api.common.autonomy.ProcessExecutionFailedException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -52,7 +46,7 @@ /** * Unit tests for {@link AgenticCommandHandlerFunctionAdapter}, verifying the runtime - * agent resolution by aggregate name and the {@link Autonomy} fallback when no matching + * agent resolution by aggregate name and the default agent fallback when no matching * agent is found. */ @ExtendWith(MockitoExtension.class) @@ -93,12 +87,6 @@ public String getAggregateId() { @Mock private Blackboard blackboard; - @Mock - private PlatformServices platformServices; - - @Mock - private Autonomy autonomy; - private final CommandType commandType = new CommandType<>("TestCommand", 1, TestCommand.class, false, false, false); @@ -157,20 +145,17 @@ void shouldResolveAgentBySuffixNameMatch() { } // ------------------------------------------------------------------------- - // Autonomy fallback + // Default agent fallback // ------------------------------------------------------------------------- @Test - void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException { - when(agentPlatform.agents()).thenReturn(List.of()); - when(agentPlatform.getPlatformServices()).thenReturn(platformServices); - when(platformServices.autonomy()).thenReturn(autonomy); - - AgentProcessExecution execution = mock(AgentProcessExecution.class); - when(execution.getAgentProcess()).thenReturn(agentProcess); - when(autonomy.chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) - .thenReturn(execution); + void shouldUseDefaultAgentWhenNoNameMatch() { + Agent defaultAgent = mock(Agent.class); + when(defaultAgent.getName()).thenReturn("SomeOtherAgent"); + when(agentPlatform.agents()).thenReturn(List.of(defaultAgent)); + when(agentPlatform.createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); when(agentProcess.getStatus()).thenReturn(null); var adapter = new AgenticCommandHandlerFunctionAdapter<>( @@ -180,21 +165,12 @@ void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException Stream result = adapter.apply(new TestCommand("agg-1"), new TestState("agg-1")); assertThat(result).isEmpty(); - verify(autonomy).chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); - verify(agentPlatform, never()).createAgentProcess(any(), any(), any(Map.class)); + verify(agentPlatform).createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class)); } @Test - void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutionException { + void shouldThrowWhenNoAgentsDeployed() { when(agentPlatform.agents()).thenReturn(List.of()); - when(agentPlatform.getPlatformServices()).thenReturn(platformServices); - when(platformServices.autonomy()).thenReturn(autonomy); - - AgentProcess failedProcess = mock(AgentProcess.class); - doThrow(new ProcessExecutionFailedException(failedProcess, "Autonomy failed")) - .when(autonomy).chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); var adapter = new AgenticCommandHandlerFunctionAdapter<>( aggregate, "UnknownAggregate", commandType, agentPlatform, @@ -202,8 +178,7 @@ void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutio assertThatThrownBy(() -> adapter.apply(new TestCommand("agg-1"), new TestState("agg-1"))) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("Autonomy fallback failed") - .hasMessageContaining("TestCommand") + .hasMessageContaining("No agents are deployed") .hasMessageContaining("UnknownAggregate"); } @@ -239,21 +214,23 @@ void shouldPreferExactMatchOverSuffixMatch() { // ------------------------------------------------------------------------- @Test - void resolveAgentByNameShouldReturnEmptyWhenNoAgentsDeployed() { + void resolveAgentByNameShouldThrowWhenNoAgentsDeployed() { when(agentPlatform.agents()).thenReturn(List.of()); - assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) - .isEmpty(); + assertThatThrownBy(() -> + AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("No agents are deployed"); } @Test - void resolveAgentByNameShouldReturnEmptyWhenNoNameMatches() { + void resolveAgentByNameShouldReturnDefaultWhenNoNameMatches() { Agent agent = mock(Agent.class); when(agent.getName()).thenReturn("SomeOtherAgent"); when(agentPlatform.agents()).thenReturn(List.of(agent)); assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) - .isEmpty(); + .isSameAs(agent); } @Test @@ -263,7 +240,7 @@ void resolveAgentByNameShouldFindExactMatch() { when(agentPlatform.agents()).thenReturn(List.of(agent)); assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Wallet")) - .contains(agent); + .isSameAs(agent); } @Test @@ -273,7 +250,7 @@ void resolveAgentByNameShouldFindSuffixMatch() { when(agentPlatform.agents()).thenReturn(List.of(agent)); assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Wallet")) - .contains(agent); + .isSameAs(agent); } // ------------------------------------------------------------------------- diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java index 96dd0910..64d02ccb 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java @@ -17,12 +17,6 @@ package org.elasticsoftware.akces.agentic.runtime; -import com.embabel.agent.api.common.PlatformServices; -import com.embabel.agent.api.common.autonomy.AgentProcessExecution; -import com.embabel.agent.api.common.autonomy.Autonomy; -import com.embabel.agent.api.common.autonomy.GoalChoiceApprover; -import com.embabel.agent.api.common.autonomy.ProcessExecutionException; -import com.embabel.agent.api.common.autonomy.ProcessExecutionFailedException; import com.embabel.agent.core.Agent; import com.embabel.agent.core.AgentPlatform; import com.embabel.agent.core.AgentProcess; @@ -50,7 +44,7 @@ /** * Unit tests for {@link AgenticEventHandlerFunctionAdapter}, verifying the runtime - * agent resolution by aggregate name and the {@link Autonomy} fallback when no matching + * agent resolution by aggregate name and the default agent fallback when no matching * agent is found. */ @ExtendWith(MockitoExtension.class) @@ -83,12 +77,6 @@ public String getAggregateId() { @Mock private Blackboard blackboard; - @Mock - private PlatformServices platformServices; - - @Mock - private Autonomy autonomy; - private final DomainEventType eventType = new DomainEventType<>("ExternalEvent", 1, TestExternalEvent.class, false, true, false, false); @@ -149,20 +137,17 @@ void shouldResolveAgentBySuffixNameMatch() { } // ------------------------------------------------------------------------- - // Autonomy fallback + // Default agent fallback // ------------------------------------------------------------------------- @Test - void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException { - when(agentPlatform.agents()).thenReturn(List.of()); - when(agentPlatform.getPlatformServices()).thenReturn(platformServices); - when(platformServices.autonomy()).thenReturn(autonomy); - - AgentProcessExecution execution = mock(AgentProcessExecution.class); - when(execution.getAgentProcess()).thenReturn(agentProcess); - when(autonomy.chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class))) - .thenReturn(execution); + void shouldUseDefaultAgentWhenNoNameMatch() { + Agent defaultAgent = mock(Agent.class); + when(defaultAgent.getName()).thenReturn("SomeOtherAgent"); + when(agentPlatform.agents()).thenReturn(List.of(defaultAgent)); + when(agentPlatform.createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class))) + .thenReturn(agentProcess); + when(agentProcess.getFinished()).thenReturn(true); when(agentProcess.getStatus()).thenReturn(null); var adapter = new AgenticEventHandlerFunctionAdapter<>( @@ -173,21 +158,12 @@ void shouldFallBackToAutonomyWhenNoAgentFound() throws ProcessExecutionException new TestExternalEvent("agg-1"), new TestState("agg-1")); assertThat(result).isEmpty(); - verify(autonomy).chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); - verify(agentPlatform, never()).createAgentProcess(any(), any(), any(Map.class)); + verify(agentPlatform).createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class)); } @Test - void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutionException { + void shouldThrowWhenNoAgentsDeployed() { when(agentPlatform.agents()).thenReturn(List.of()); - when(agentPlatform.getPlatformServices()).thenReturn(platformServices); - when(platformServices.autonomy()).thenReturn(autonomy); - - AgentProcess failedProcess = mock(AgentProcess.class); - doThrow(new ProcessExecutionFailedException(failedProcess, "Autonomy failed")) - .when(autonomy).chooseAndAccomplishGoal( - any(GoalChoiceApprover.class), eq(agentPlatform), any(Map.class)); var adapter = new AgenticEventHandlerFunctionAdapter<>( aggregate, "UnknownAggregate", eventType, agentPlatform, @@ -196,8 +172,7 @@ void shouldWrapAutonomyExceptionInIllegalStateException() throws ProcessExecutio assertThatThrownBy(() -> adapter.apply( new TestExternalEvent("agg-1"), new TestState("agg-1"))) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("Autonomy fallback failed") - .hasMessageContaining("ExternalEvent") + .hasMessageContaining("No agents are deployed") .hasMessageContaining("UnknownAggregate"); } From 9749406a0825cb71ddb99b202e557fecc7ad25a8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 13:30:37 +0000 Subject: [PATCH 09/10] Add DefaultAgent in o.e.akces.agentic.embabel package Create a default Embabel agent annotated with @Agent(planner = UTILITY) that will always be present on the AgentPlatform. This serves as the fallback agent when no other agent matches the aggregate name during agent resolution in the agentic command and event handler adapters. Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/1a5b2b84-6e16-4eca-9184-2c6f843ddd89 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akces/agentic/embabel/DefaultAgent.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java 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 new file mode 100644 index 00000000..efd4d642 --- /dev/null +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/embabel/DefaultAgent.java @@ -0,0 +1,33 @@ +/* + * 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.Agent; +import com.embabel.agent.api.common.PlannerType; + +/** + * The default Embabel agent that is always present on the {@link com.embabel.agent.core.AgentPlatform}. + * This agent is used as a fallback when no other agent matches the aggregate name during + * agent resolution in the agentic command and event handler adapters. + * + *

      Uses the {@link PlannerType#UTILITY UTILITY} planner for flexible goal-based planning. + * Actions and conditions will be added as the agentic framework evolves. + */ +@Agent(description = "Default Akces agentic aggregate agent", planner = PlannerType.UTILITY) +public class DefaultAgent { +} From 3ec76500b6344e6965e41dcb57284e8e74cd26f8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Apr 2026 15:24:16 +0000 Subject: [PATCH 10/10] Use explicit DefaultAgent.AGENT_NAME for fallback instead of first available agent Updated resolveAgentByName in both adapters to look specifically for the DefaultAgent by its AGENT_NAME constant instead of using the first available agent from the platform. Set name="DefaultAgent" on the @Agent annotation. IllegalStateException is now thrown when no DefaultAgent is deployed (not just when no agents exist at all). Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/16bc6db4-5f7d-4287-b643-ff758d8cf927 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akces/agentic/embabel/DefaultAgent.java | 8 ++++- .../AgenticCommandHandlerFunctionAdapter.java | 30 +++++++++++-------- .../AgenticEventHandlerFunctionAdapter.java | 22 ++++++++------ ...nticCommandHandlerFunctionAdapterTest.java | 27 +++++++++++++---- ...genticEventHandlerFunctionAdapterTest.java | 22 ++++++++++++-- 5 files changed, 79 insertions(+), 30 deletions(-) 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 efd4d642..7d2c1618 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 @@ -28,6 +28,12 @@ *

      Uses the {@link PlannerType#UTILITY UTILITY} planner for flexible goal-based planning. * Actions and conditions will be added as the agentic framework evolves. */ -@Agent(description = "Default Akces agentic aggregate agent", planner = PlannerType.UTILITY) +@Agent(name = DefaultAgent.AGENT_NAME, description = "Default Akces agentic aggregate agent", planner = PlannerType.UTILITY) 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"; } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java index 021661f1..0e2dd539 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapter.java @@ -22,6 +22,7 @@ import com.embabel.agent.core.AgentProcess; import com.embabel.agent.core.ProcessOptions; import jakarta.annotation.Nonnull; +import org.elasticsoftware.akces.agentic.embabel.DefaultAgent; import org.elasticsoftware.akces.aggregate.*; import org.elasticsoftware.akces.commands.Command; import org.elasticsoftware.akces.control.AggregateServiceRecord; @@ -48,8 +49,8 @@ * memories, aggregate service records, and condition flags. *

    18. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching * the aggregate name against deployed agent names (exact match or - * {@code {aggregateName}Agent} suffix match). If no match is found, the first - * available agent from the platform is used as a default.
      19. + * {@code {aggregateName}Agent} suffix match). If no match is found, the + * {@link DefaultAgent} is used as a fallback. *
    19. Creates an {@link AgentProcess} via * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls * {@link AgentProcess#tick()} in a loop until the process reaches an end state @@ -217,25 +218,26 @@ private List> getAllRegisteredEventTypes() { /** * Resolves an {@link Agent} from the {@link AgentPlatform} by matching deployed agent - * names against the given aggregate name. If no match is found, the first available - * agent from the platform is returned as a default. + * names against the given aggregate name. If no match is found, the + * {@link DefaultAgent} is returned as a fallback. * *

      Matching rules (checked in order): *

        *
      1. Exact match: {@code agent.getName().equals(aggregateName)}
        2. *
      2. Agent-suffix match: {@code agent.getName().equals(aggregateName + "Agent")}
        3. - *
      3. Default: first available agent from the platform
        4. + *
      4. Default: the {@link DefaultAgent} identified by {@link DefaultAgent#AGENT_NAME}
        5. *
      * * @param platform the agent platform containing deployed agents * @param aggregateName the aggregate name to match against * @return the matched or default agent; never {@code null} - * @throws IllegalStateException if no agents are deployed on the platform + * @throws IllegalStateException if the {@link DefaultAgent} is not deployed on the platform */ static Agent resolveAgentByName(AgentPlatform platform, String aggregateName) { String suffixName = aggregateName + "Agent"; Collection agents = platform.agents(); Agent suffixMatch = null; + Agent defaultAgentMatch = null; for (Agent agent : agents) { String agentName = agent.getName(); @@ -245,23 +247,25 @@ static Agent resolveAgentByName(AgentPlatform platform, String aggregateName) { if (suffixMatch == null && agentName.equals(suffixName)) { suffixMatch = agent; } + if (defaultAgentMatch == null && agentName.equals(DefaultAgent.AGENT_NAME)) { + defaultAgentMatch = agent; + } } if (suffixMatch != null) { return suffixMatch; } - // No name match — use the first available agent as default - if (!agents.isEmpty()) { - Agent defaultAgent = agents.iterator().next(); + // No name match — fall back to the DefaultAgent + if (defaultAgentMatch != null) { logger.info("No agent found matching aggregate name '{}'; using default agent '{}'", - aggregateName, defaultAgent.getName()); - return defaultAgent; + aggregateName, defaultAgentMatch.getName()); + return defaultAgentMatch; } throw new IllegalStateException( - "No agents are deployed on the AgentPlatform. " - + "At least one agent must be available to handle aggregate '" + "No DefaultAgent ('" + DefaultAgent.AGENT_NAME + "') is deployed on the AgentPlatform. " + + "At least the DefaultAgent must be available to handle aggregate '" + aggregateName + "'."); } diff --git a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java index ba62f465..c096c57a 100644 --- a/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java +++ b/main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapter.java @@ -22,6 +22,7 @@ import com.embabel.agent.core.AgentProcess; import com.embabel.agent.core.ProcessOptions; import jakarta.annotation.Nonnull; +import org.elasticsoftware.akces.agentic.embabel.DefaultAgent; import org.elasticsoftware.akces.aggregate.*; import org.elasticsoftware.akces.control.AggregateServiceRecord; import org.elasticsoftware.akces.events.DomainEvent; @@ -47,8 +48,8 @@ * memories, aggregate service records, and condition flags.
      20. *
    20. Attempts to resolve an {@link Agent} from the {@link AgentPlatform} by matching * the aggregate name against deployed agent names (exact match or - * {@code {aggregateName}Agent} suffix match). If no match is found, the first - * available agent from the platform is used as a default.
      21. + * {@code {aggregateName}Agent} suffix match). If no match is found, the + * {@link DefaultAgent} is used as a fallback. *
    21. Creates an {@link AgentProcess} via * {@link AgentPlatform#createAgentProcess(Agent, ProcessOptions, Map)} and calls * {@link AgentProcess#tick()} in a loop until the process reaches an end state @@ -138,6 +139,7 @@ private static Agent resolveAgentByName(AgentPlatform agentPlatform, String aggr String suffixName = aggregateName + "Agent"; Collection agents = agentPlatform.agents(); Agent suffixMatch = null; + Agent defaultAgentMatch = null; for (Agent agent : agents) { String agentName = agent.getName(); @@ -147,23 +149,25 @@ private static Agent resolveAgentByName(AgentPlatform agentPlatform, String aggr if (suffixMatch == null && agentName.equals(suffixName)) { suffixMatch = agent; } + if (defaultAgentMatch == null && agentName.equals(DefaultAgent.AGENT_NAME)) { + defaultAgentMatch = agent; + } } if (suffixMatch != null) { return suffixMatch; } - // No name match — use the first available agent as default - if (!agents.isEmpty()) { - Agent defaultAgent = agents.iterator().next(); + // No name match — fall back to the DefaultAgent + if (defaultAgentMatch != null) { logger.info("No agent found matching aggregate name '{}'; using default agent '{}'", - aggregateName, defaultAgent.getName()); - return defaultAgent; + aggregateName, defaultAgentMatch.getName()); + return defaultAgentMatch; } throw new IllegalStateException( - "No agents are deployed on the AgentPlatform. " - + "At least one agent must be available to handle aggregate '" + "No DefaultAgent ('" + DefaultAgent.AGENT_NAME + "') is deployed on the AgentPlatform. " + + "At least the DefaultAgent must be available to handle aggregate '" + aggregateName + "'."); } diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java index 79523ad1..33e6dd76 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticCommandHandlerFunctionAdapterTest.java @@ -22,6 +22,7 @@ import com.embabel.agent.core.AgentProcess; import com.embabel.agent.core.Blackboard; import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.agentic.embabel.DefaultAgent; import org.elasticsoftware.akces.aggregate.*; import org.elasticsoftware.akces.annotations.CommandInfo; import org.elasticsoftware.akces.annotations.DomainEventInfo; @@ -151,7 +152,7 @@ void shouldResolveAgentBySuffixNameMatch() { @Test void shouldUseDefaultAgentWhenNoNameMatch() { Agent defaultAgent = mock(Agent.class); - when(defaultAgent.getName()).thenReturn("SomeOtherAgent"); + when(defaultAgent.getName()).thenReturn(DefaultAgent.AGENT_NAME); when(agentPlatform.agents()).thenReturn(List.of(defaultAgent)); when(agentPlatform.createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class))) .thenReturn(agentProcess); @@ -168,6 +169,22 @@ void shouldUseDefaultAgentWhenNoNameMatch() { verify(agentPlatform).createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class)); } + @Test + void shouldThrowWhenNoDefaultAgentDeployed() { + Agent otherAgent = mock(Agent.class); + when(otherAgent.getName()).thenReturn("SomeOtherAgent"); + when(agentPlatform.agents()).thenReturn(List.of(otherAgent)); + + var adapter = new AgenticCommandHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", commandType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThatThrownBy(() -> adapter.apply(new TestCommand("agg-1"), new TestState("agg-1"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("No DefaultAgent") + .hasMessageContaining("UnknownAggregate"); + } + @Test void shouldThrowWhenNoAgentsDeployed() { when(agentPlatform.agents()).thenReturn(List.of()); @@ -178,7 +195,7 @@ void shouldThrowWhenNoAgentsDeployed() { assertThatThrownBy(() -> adapter.apply(new TestCommand("agg-1"), new TestState("agg-1"))) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("No agents are deployed") + .hasMessageContaining("No DefaultAgent") .hasMessageContaining("UnknownAggregate"); } @@ -220,13 +237,13 @@ void resolveAgentByNameShouldThrowWhenNoAgentsDeployed() { assertThatThrownBy(() -> AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("No agents are deployed"); + .hasMessageContaining("No DefaultAgent"); } @Test - void resolveAgentByNameShouldReturnDefaultWhenNoNameMatches() { + void resolveAgentByNameShouldReturnDefaultAgentWhenNoNameMatches() { Agent agent = mock(Agent.class); - when(agent.getName()).thenReturn("SomeOtherAgent"); + when(agent.getName()).thenReturn(DefaultAgent.AGENT_NAME); when(agentPlatform.agents()).thenReturn(List.of(agent)); assertThat(AgenticCommandHandlerFunctionAdapter.resolveAgentByName(agentPlatform, "Test")) diff --git a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java index 64d02ccb..4f177bf2 100644 --- a/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java +++ b/main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticEventHandlerFunctionAdapterTest.java @@ -22,6 +22,7 @@ import com.embabel.agent.core.AgentProcess; import com.embabel.agent.core.Blackboard; import com.embabel.agent.core.ProcessOptions; +import org.elasticsoftware.akces.agentic.embabel.DefaultAgent; import org.elasticsoftware.akces.aggregate.*; import org.elasticsoftware.akces.annotations.DomainEventInfo; import org.elasticsoftware.akces.events.DomainEvent; @@ -143,7 +144,7 @@ void shouldResolveAgentBySuffixNameMatch() { @Test void shouldUseDefaultAgentWhenNoNameMatch() { Agent defaultAgent = mock(Agent.class); - when(defaultAgent.getName()).thenReturn("SomeOtherAgent"); + when(defaultAgent.getName()).thenReturn(DefaultAgent.AGENT_NAME); when(agentPlatform.agents()).thenReturn(List.of(defaultAgent)); when(agentPlatform.createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class))) .thenReturn(agentProcess); @@ -161,6 +162,23 @@ void shouldUseDefaultAgentWhenNoNameMatch() { verify(agentPlatform).createAgentProcess(eq(defaultAgent), eq(ProcessOptions.DEFAULT), any(Map.class)); } + @Test + void shouldThrowWhenNoDefaultAgentDeployed() { + Agent otherAgent = mock(Agent.class); + when(otherAgent.getName()).thenReturn("SomeOtherAgent"); + when(agentPlatform.agents()).thenReturn(List.of(otherAgent)); + + var adapter = new AgenticEventHandlerFunctionAdapter<>( + aggregate, "UnknownAggregate", eventType, agentPlatform, + List.of(), List.of(), Collections::emptyList); + + assertThatThrownBy(() -> adapter.apply( + new TestExternalEvent("agg-1"), new TestState("agg-1"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("No DefaultAgent") + .hasMessageContaining("UnknownAggregate"); + } + @Test void shouldThrowWhenNoAgentsDeployed() { when(agentPlatform.agents()).thenReturn(List.of()); @@ -172,7 +190,7 @@ void shouldThrowWhenNoAgentsDeployed() { assertThatThrownBy(() -> adapter.apply( new TestExternalEvent("agg-1"), new TestState("agg-1"))) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("No agents are deployed") + .hasMessageContaining("No DefaultAgent") .hasMessageContaining("UnknownAggregate"); }