Add auto-create hook for AgenticAggregate singleton state initialization

main/agentic/src/main/java/org/elasticsoftware/akces/agentic/AgenticAggregateRuntime.java

@@ -26,11 +26,16 @@
 import org.elasticsoftware.akces.aggregate.AggregateRuntime;
 import org.elasticsoftware.akces.aggregate.CommandType;
 import org.elasticsoftware.akces.aggregate.DomainEventType;
+import org.elasticsoftware.akces.aggregate.IndexParams;
 import org.elasticsoftware.akces.aggregate.MemoryAwareState;
 import org.elasticsoftware.akces.protocol.AggregateStateRecord;
+import org.elasticsoftware.akces.protocol.DomainEventRecord;
+import org.elasticsoftware.akces.protocol.ProtocolRecord;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 /**
  * Extended runtime interface for {@link org.elasticsoftware.akces.aggregate.AgenticAggregate}s.
@@ -89,4 +94,22 @@ public interface AgenticAggregateRuntime extends AggregateRuntime {
      * @throws IOException if deserialization fails
      */
     List<AgenticAggregateMemory> getMemories(AggregateStateRecord stateRecord) throws IOException;
+
+    /**
+     * Initializes the singleton aggregate state by invoking the aggregate's
+     * {@link org.elasticsoftware.akces.aggregate.AgenticAggregate#getCreateDomainEvent()
+     * getCreateDomainEvent()} hook and applying the resulting event through the event-sourcing
+     * create handler.
+     *
+     * <p>This method is called by the partition when it detects that no state exists yet
+     * for the agentic aggregate. The produced {@link AggregateStateRecord} and
+     * {@link DomainEventRecord} are written to the Kafka topics via the provided consumers.
+     *
+     * @param protocolRecordConsumer consumer for the produced state and domain-event records
+     * @param domainEventIndexer     indexer callback for optional secondary indexing
+     * @throws IOException if serialisation fails
+     */
+    void initializeState(Consumer<ProtocolRecord> protocolRecordConsumer,
+                         BiConsumer<DomainEventRecord, IndexParams> domainEventIndexer)
+            throws IOException;
 }

main/agentic/src/main/java/org/elasticsoftware/akces/agentic/beans/AgenticAggregateRuntimeFactory.java

@@ -132,7 +132,7 @@ public AgenticAggregateRuntime getObject() {
 
         KafkaAggregateRuntime kafkaRuntime = createRuntime(agenticInfo, aggregate, agentPlatform);
         return new KafkaAgenticAggregateRuntime(
-                kafkaRuntime, objectMapper, agenticInfo.stateClass(), agentPlatform);
+                kafkaRuntime, objectMapper, agenticInfo.stateClass(), agentPlatform, aggregate);
     }
 
     @Override

main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/AgenticAggregatePartition.java

@@ -64,6 +64,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.elasticsoftware.akces.kafka.AggregatePartitionState.INITIALIZING;
+import static org.elasticsoftware.akces.kafka.AggregatePartitionState.INITIALIZING_STATE;
 import static org.elasticsoftware.akces.kafka.AggregatePartitionState.LOADING_STATE;
 import static org.elasticsoftware.akces.kafka.AggregatePartitionState.PROCESSING;
 import static org.elasticsoftware.akces.kafka.AggregatePartitionState.SHUTTING_DOWN;
@@ -285,10 +286,11 @@ private void process() {
                 }
                 Map<TopicPartition, Long> endOffsets = consumer.endOffsets(List.of(statePartition));
                 if (endOffsets.getOrDefault(statePartition, 0L) == 0L) {
-                    // No existing state — start processing immediately
-                    logger.info("No existing state for {}Aggregate AgenticPartition, starting PROCESSING",
-                            runtime.getName());
-                    processState = PROCESSING;
+                    // No existing state — transition to INITIALIZING_STATE to auto-create
+                    // the singleton aggregate state in the next poll cycle.
+                    logger.info("No existing state for {}Aggregate AgenticPartition, " +
+                            "switching to INITIALIZING_STATE", runtime.getName());
+                    processState = INITIALIZING_STATE;
                 } else {
                     // Pause command/event/external topics while loading state
                     List<TopicPartition> toPause = new ArrayList<>(
@@ -297,6 +299,28 @@ private void process() {
                     consumer.pause(toPause);
                     processState = LOADING_STATE;
                 }
+            } else if (processState == INITIALIZING_STATE) {
+                logger.info("Auto-creating initial state for {}Aggregate AgenticPartition",
+                        runtime.getName());
+                try {
+                    producer.beginTransaction();
+                    runtime.initializeState(this::send, this::index);
+                    producer.commitTransaction();
+                    stateRepository.commit();
+                    processState = PROCESSING;
+                    logger.info("Successfully auto-created initial state for {}Aggregate, " +
+                            "switching to PROCESSING", runtime.getName());
+                } catch (Exception e) {
+                    logger.error("Failed to auto-create initial state for {}Aggregate — " +
+                            "shutting down", runtime.getName(), e);
+                    try {
+                        producer.abortTransaction();
+                    } catch (KafkaException ae) {
+                        logger.error("Failed to abort transaction", ae);
+                    }
+                    stateRepository.rollback();
+                    processState = SHUTTING_DOWN;
+                }
             }
         } catch (WakeupException | InterruptException ignore) {
             // Non-fatal; ignore

main/agentic/src/main/java/org/elasticsoftware/akces/agentic/runtime/KafkaAgenticAggregateRuntime.java

@@ -32,6 +32,7 @@
 import org.elasticsoftware.akces.aggregate.*;
 import org.elasticsoftware.akces.commands.Command;
 import org.elasticsoftware.akces.commands.CommandBus;
+import org.elasticsoftware.akces.kafka.KafkaAggregateRuntime;
 import org.elasticsoftware.akces.protocol.AggregateStateRecord;
 import org.elasticsoftware.akces.protocol.CommandRecord;
 import org.elasticsoftware.akces.protocol.DomainEventRecord;
@@ -57,19 +58,21 @@
 /**
  * Kafka-backed implementation of {@link AgenticAggregateRuntime}.
  *
- * <p>Wraps a {@link AggregateRuntime} delegate (typically a {@code KafkaAggregateRuntime}) and
- * adds the memory-aware {@link #getMemories(AggregateStateRecord)} method. All other
+ * <p>Wraps a {@link KafkaAggregateRuntime} delegate and adds the memory-aware
+ * {@link #getMemories(AggregateStateRecord)} method as well as the auto-create
+ * {@link #initializeState(Consumer, BiConsumer)} method. All other
  * {@link AggregateRuntime} operations are forwarded to the delegate.
  */
 public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime {
 
     private static final Logger logger =
             LoggerFactory.getLogger(KafkaAgenticAggregateRuntime.class);
 
-    private final AggregateRuntime delegate;
+    private final KafkaAggregateRuntime delegate;
     private final ObjectMapper objectMapper;
     private final Class<? extends AggregateState> stateClass;
     private final AgentPlatform agentPlatform;
+    private final AgenticAggregate<?> aggregate;
 
     /**
      * Creates a new {@code KafkaAgenticAggregateRuntime}.
@@ -79,15 +82,20 @@ public class KafkaAgenticAggregateRuntime implements AgenticAggregateRuntime {
      * @param stateClass    the concrete state class used by this aggregate
      * @param agentPlatform the Embabel {@link AgentPlatform} used for AI-assisted processing;
      *                      must not be {@code null}
+     * @param aggregate     the agentic aggregate instance whose
+     *                      {@link AgenticAggregate#getCreateDomainEvent()} method provides the
+     *                      auto-create event
      */
-    public KafkaAgenticAggregateRuntime(AggregateRuntime delegate,
+    public KafkaAgenticAggregateRuntime(KafkaAggregateRuntime delegate,
                                         ObjectMapper objectMapper,
                                         Class<? extends AggregateState> stateClass,
-                                        AgentPlatform agentPlatform) {
+                                        AgentPlatform agentPlatform,
+                                        AgenticAggregate<?> aggregate) {
         this.delegate = Objects.requireNonNull(delegate, "delegate must not be null");
         this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null");
         this.stateClass = Objects.requireNonNull(stateClass, "stateClass must not be null");
         this.agentPlatform = Objects.requireNonNull(agentPlatform, "agentPlatform must not be null");
+        this.aggregate = Objects.requireNonNull(aggregate, "aggregate must not be null");
     }
 
     // -------------------------------------------------------------------------
@@ -121,6 +129,26 @@ public List<AgenticAggregateMemory> getMemories(AggregateStateRecord stateRecord
         return List.of();
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Calls {@link AgenticAggregate#getCreateDomainEvent()} on the aggregate instance and
+     * delegates to the underlying {@link KafkaAggregateRuntime#handleAutoCreateDomainEvent}
+     * to apply the event-sourcing create handler and produce the initial state and event
+     * records.
+     */
+    @Override
+    public void initializeState(Consumer<ProtocolRecord> protocolRecordConsumer,
+                                BiConsumer<DomainEventRecord, IndexParams> domainEventIndexer)
+            throws IOException {
+        DomainEvent createEvent = aggregate.getCreateDomainEvent();
+        Objects.requireNonNull(createEvent,
+                "AgenticAggregate.getCreateDomainEvent() must not return null");
+        logger.info("Auto-creating initial state for {}Aggregate using {}",
+                getName(), createEvent.getClass().getSimpleName());
+        delegate.handleAutoCreateDomainEvent(createEvent, protocolRecordConsumer, domainEventIndexer);
+    }
+
     // -------------------------------------------------------------------------
     // AggregateRuntime delegation
     // -------------------------------------------------------------------------