Add auto-create hook for AgenticAggregate singleton state initialization by Copilot · Pull Request #332 · elasticsoftwarefoundation/akces-framework

Copilot · 2026-04-11T09:19:24Z

AgenticAggregate is a singleton (one instance per partition) but had no mechanism to automatically bootstrap its initial state—it required an external command to trigger creation. This adds a framework-level hook that auto-creates state when the partition detects none exists.

API (`AgenticAggregate` interface)

Added abstract DomainEvent getCreateDomainEvent() — the hook implementors provide to supply the create event for their singleton aggregate

Runtime (`KafkaAggregateRuntime`)

Added handleAutoCreateDomainEvent(DomainEvent, Consumer<ProtocolRecord>, BiConsumer<DomainEventRecord, IndexParams>) — same flow as handleCreateCommand but without command materialization
Added INITIALIZING_STATE to AggregatePartitionState enum for the dedicated auto-create phase

Agentic module

AgenticAggregateRuntime — added initializeState() interface method
KafkaAgenticAggregateRuntime — stores AgenticAggregate<?> ref, changed delegate type to KafkaAggregateRuntime, implements initializeState() by calling aggregate.getCreateDomainEvent() then delegating
AgenticAggregatePartition — during INITIALIZING, when endOffsets == 0L, transitions to INITIALIZING_STATE. The INITIALIZING_STATE phase opens a Kafka transaction, calls runtime.initializeState(), and on success transitions to PROCESSING; on failure (catching broad Exception including unchecked SerializationException) aborts the transaction and transitions to SHUTTING_DOWN to fail fast
AgenticAggregateRuntimeFactory — passes aggregate instance to KafkaAgenticAggregateRuntime constructor

Usage

@AgenticAggregateInfo(value = "MyBot", stateClass = MyBotState.class)
public class MyBot implements AgenticAggregate<MyBotState> {

    @Override
    public DomainEvent getCreateDomainEvent() {
        return new MyBotCreatedEvent(getName());
    }

    @EventSourcingHandler(create = true)
    public MyBotState create(MyBotCreatedEvent event, MyBotState isNull) {
        return new MyBotState(event.botId(), List.of());
    }
    // ...
}

Test

AgenticAggregateAutoCreateTest — complete TestBot aggregate implementation exercising the hook contract, event sourcing handler, full create flow, initializeState() delegation, and null-safety. Existing tests updated for the new getCreateDomainEvent() contract.

- Add getCreateDomainEvent() to AgenticAggregate interface - Add handleAutoCreateDomainEvent() to KafkaAggregateRuntime - Add initializeState() to AgenticAggregateRuntime interface - Implement initializeState() in KafkaAgenticAggregateRuntime - Update AgenticAggregateRuntimeFactory to pass aggregate reference - Hook auto-create into AgenticAggregatePartition during INITIALIZING Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/040fd779-8f79-4ca7-a05f-5788f1dfcf04 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

…enticAggregateAutoCreateTest - Update existing AgenticAggregate test implementations to implement getCreateDomainEvent() - Update KafkaAgenticAggregateRuntimeTest for new constructor signature - Add comprehensive AgenticAggregateAutoCreateTest with TestBot aggregate implementation - All 87 agentic tests and 54 api tests pass Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/040fd779-8f79-4ca7-a05f-5788f1dfcf04 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

Copilot

Pull request overview

Adds a framework-level auto-create hook so agentic singleton aggregates can bootstrap initial state automatically when a partition starts with no state, instead of requiring an external create command.

Changes:

Introduces AgenticAggregate#getCreateDomainEvent() and updates docs/tests to satisfy the new contract.
Adds KafkaAggregateRuntime#handleAutoCreateDomainEvent(...) and wires initializeState() through KafkaAgenticAggregateRuntime.
Updates AgenticAggregatePartition to auto-create state during INITIALIZING when the state topic is empty, plus adds/updates tests.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java	Adds `handleAutoCreateDomainEvent(...)` to create initial state from a domain event without command materialization.
main/api/src/main/java/org/elasticsoftware/akces/aggregate/AgenticAggregate.java	Adds required `getCreateDomainEvent()` hook for singleton bootstrap.
main/api/src/main/java/org/elasticsoftware/akces/annotations/AgenticAggregateInfo.java	Updates annotation docs to reflect event-based singleton creation contract.
main/agentic/src/main/java/org/elasticsoftware/akces/agentic/AgenticAggregateRuntime.java	Adds `initializeState(...)` API for partitions to trigger auto-create.
main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java	Stores aggregate reference and delegates `initializeState()` to `KafkaAggregateRuntime#handleAutoCreateDomainEvent`.
main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregatePartition.java	Implements the INITIALIZING-time auto-create transactional flow when no state exists.
main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java	Passes the aggregate instance into `KafkaAgenticAggregateRuntime`.
main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java	Adds end-to-end unit test coverage for the new hook and delegation behavior.
main/agentic/src/test/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntimeTest.java	Updates constructor signature expectations and adds null-aggregate validation.
main/api/src/test/java/org/elasticsoftware/akces/aggregate/AgenticAggregateInfoTest.java	Updates test aggregates to implement `getCreateDomainEvent()`.
main/api/src/test/java/org/elasticsoftware/akces/aggregate/MemoryAwareStateTest.java	Updates test aggregates to implement `getCreateDomainEvent()`.
main/eventcatalog/src/test/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessorTest.java	Updates embedded agentic aggregate example to implement `getCreateDomainEvent()`.

Copilot · 2026-04-11T09:23:53Z

+                        }
+                        stateRepository.rollback();
+                    }
                    processState = PROCESSING;
                } else {


In INITIALIZING, processState is set to PROCESSING unconditionally even if initializeState() fails and the transaction is aborted/rolled back. This can leave the partition running without any state (subsequent commands will see missing state) and also hides the initialization failure. Only transition to PROCESSING after a successful auto-create; on failure either keep INITIALIZING (with retry/backoff) or switch to SHUTTING_DOWN to fail fast.

create a new ProcessState for INITIALIZING_STATE and implement the State initialization logic there

Added INITIALIZING_STATE to AggregatePartitionState enum. The INITIALIZING phase now transitions to INITIALIZING_STATE when no state exists, and the auto-create logic runs in the dedicated INITIALIZING_STATE phase. On success → PROCESSING, on failure → SHUTTING_DOWN (fail-fast). Also catches broad Exception to handle unchecked SerializationException. See commit a52240f.

Copilot · 2026-04-11T09:23:53Z

+    void initializeState(java.util.function.Consumer<org.elasticsoftware.akces.protocol.ProtocolRecord> protocolRecordConsumer,
+                         java.util.function.BiConsumer<org.elasticsoftware.akces.protocol.DomainEventRecord, org.elasticsoftware.akces.aggregate.IndexParams> domainEventIndexer)
+            throws IOException;


initializeState(...) is declared with fully-qualified parameter types, unlike the rest of this interface which uses imports. This makes the public API harder to read and is inconsistent with the surrounding style. Prefer importing ProtocolRecord, DomainEventRecord, IndexParams, Consumer, and BiConsumer and using those simple names in the method signature.

Fixed — added proper imports for Consumer, BiConsumer, ProtocolRecord, DomainEventRecord, and IndexParams, replacing the fully-qualified types. See commit a52240f.

Copilot · 2026-04-11T09:23:54Z

+                    try {
+                        producer.beginTransaction();
+                        runtime.initializeState(this::send, this::index);
+                        producer.commitTransaction();
+                        stateRepository.commit();


The auto-create transaction only catches KafkaException and checked IOException. initializeState() / serialization can also throw unchecked exceptions (e.g., org.apache.kafka.common.errors.SerializationException), which would bypass the abort/rollback path and leave the producer transaction open until the partition thread dies. Consider catching Exception (or at least RuntimeException) around the transactional block and always aborting + rolling back when the transaction isn’t committed.

jwijgerd · 2026-04-11T09:27:24Z

+                        }
+                        stateRepository.rollback();
+                    }
                    processState = PROCESSING;
                } else {


create a new ProcessState for INITIALIZING_STATE and implement the State initialization logic there

jwijgerd · 2026-04-11T09:27:43Z

+    void initializeState(java.util.function.Consumer<org.elasticsoftware.akces.protocol.ProtocolRecord> protocolRecordConsumer,
+                         java.util.function.BiConsumer<org.elasticsoftware.akces.protocol.DomainEventRecord, org.elasticsoftware.akces.aggregate.IndexParams> domainEventIndexer)
+            throws IOException;


…rove error handling - Add INITIALIZING_STATE to AggregatePartitionState enum - Move auto-create logic from INITIALIZING into dedicated INITIALIZING_STATE phase - Only transition to PROCESSING on success; transition to SHUTTING_DOWN on failure - Catch broad Exception (not just KafkaException/IOException) to handle unchecked SerializationException and other runtime exceptions - Fix fully-qualified parameter types in AgenticAggregateRuntime.initializeState() to use proper imports matching the interface style Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/17027302-044b-4bba-b18f-85bef160366d Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

Copilot AI and others added 2 commits April 11, 2026 08:51

Copilot AI assigned Copilot and jwijgerd Apr 11, 2026

Copilot created this pull request from a session on behalf of jwijgerd April 11, 2026 09:19 View session

jwijgerd marked this pull request as ready for review April 11, 2026 09:19

Copilot AI review requested due to automatic review settings April 11, 2026 09:19

Copilot started reviewing on behalf of jwijgerd April 11, 2026 09:20 View session

github-code-quality Bot found potential problems Apr 11, 2026

View reviewed changes

Comment thread .../src/test/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregateAutoCreateTest.java Dismissed

Copilot AI reviewed Apr 11, 2026

View reviewed changes

jwijgerd requested changes Apr 11, 2026

View reviewed changes

Copilot started work on behalf of jwijgerd April 11, 2026 09:28 View session

Copilot finished work on behalf of jwijgerd April 11, 2026 09:39

Copilot AI requested a review from jwijgerd April 11, 2026 09:39

jwijgerd approved these changes Apr 11, 2026

View reviewed changes

jwijgerd merged commit a9eff22 into main Apr 11, 2026
7 checks passed

jwijgerd deleted the copilot/implement-agenticaggregate-create-hook branch April 11, 2026 09:50

Uh oh!

Conversation

Copilot AI commented Apr 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

API (AgenticAggregate interface)

Runtime (KafkaAggregateRuntime)

Agentic module

Usage

Test

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

jwijgerd Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

jwijgerd Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

jwijgerd Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

jwijgerd Apr 11, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Copilot AI commented Apr 11, 2026 •

edited

Loading

API (`AgenticAggregate` interface)

Runtime (`KafkaAggregateRuntime`)