diff --git a/FRAMEWORK_OVERVIEW.md b/FRAMEWORK_OVERVIEW.md new file mode 100644 index 00000000..53872648 --- /dev/null +++ b/FRAMEWORK_OVERVIEW.md @@ -0,0 +1,231 @@ +# Akces Framework — Architecture Overview + +This document provides a detailed description of the Akces framework architecture, module structure, +and core concepts. For quick-start instructions see [README.md](README.md). For Kubernetes deployment +details see [SERVICES.md](SERVICES.md). + +--- + +## Module Structure + +| Module | Artifact ID | Purpose | +|--------|-------------|---------| +| `main/api` | `akces-api` | Core interfaces, annotations, and value objects defining the programming model | +| `main/runtime` | `akces-runtime` | Kafka-backed event sourcing engine, aggregate lifecycle, state snapshots | +| `main/shared` | `akces-shared` | Serialisation, GDPR/PII encryption, schema utilities, control-plane records | +| `main/client` | `akces-client` | Command submission client and response handling | +| `main/query-support` | `akces-query-support` | Query model and database model runtime | +| `main/eventcatalog` | `akces-eventcatalog` | Compile-time annotation processor that generates EventCatalog MDX documentation | +| `main/agentic` | `akces-agentic` | AgenticAggregate support: singleton aggregates with AI-agent memory and MCP integration | + +All modules are included in the `akces-framework-bom` Bill of Materials. + +--- + +## Core Concepts + +### Aggregates + +Aggregates are the central consistency boundary in Akces. Each aggregate: + +- Is annotated with `@AggregateInfo` (name, state class, optional GDPR key generation, index name). +- Implements `Aggregate` where `S extends AggregateState`. +- Handles commands via `@CommandHandler` methods (returns `Stream`). +- Rebuilds state from events via `@EventSourcingHandler` methods (returns a new state instance). +- Is partitioned across Kafka partitions; the aggregate identifier determines the partition. + +Multiple replicas of an Aggregate service run in parallel, each owning a disjoint set of +partitions. + +### Commands and Events + +- **Commands** are `@CommandInfo`-annotated records implementing `Command`. They carry an + `@AggregateIdentifier` field used for routing. +- **Domain Events** are `@DomainEventInfo`-annotated records implementing `DomainEvent`. They are + immutable facts stored in the Kafka event log. +- Commands are validated against a generated JSON Schema before dispatch. +- The command–event lifecycle is transactional: either all events produced by a handler are + committed to Kafka, or none are. + +### Query Models and Database Models + +- **Query Models** (`@QueryModelInfo`, implements `QueryModel`) subscribe to event topics and + maintain a local RocksDB state for fast reads. +- **Database Models** (`@DatabaseModelInfo`, extends `JdbcDatabaseModel` or `JpaDatabaseModel`) + persist events into relational tables via JDBC/JPA for complex analytical queries. + +### Process Managers + +Process managers coordinate workflows that span multiple aggregates. They listen to events and +emit commands (and their own internal events) to advance a multi-step business process. + +### Schema Evolution and Upcasting + +Schemas are versioned. When a new version of a command or event is introduced, an +`@UpcastingHandler` method converts old instances to the new type so that event-sourced replays +work across schema versions. + +### GDPR / PII Data + +Fields marked with `@PIIData` are transparently encrypted using per-aggregate encryption keys +stored in a dedicated Kafka topic. Decryption happens at read time using `AkcesGDPRModule`. + +--- + +## AgenticAggregates + +### What is an AgenticAggregate? + +An **AgenticAggregate** is a special kind of aggregate designed to host an AI agent (e.g. a +Spring AI or Embabel-powered chatbot). It differs from a regular aggregate in the following ways: + +| Aspect | Regular Aggregate | AgenticAggregate | +|--------|-------------------|-----------------| +| Replicas | Multiple (Kafka-partitioned) | Always **1** (singleton) | +| Partition count | Configurable | Fixed at **1** per topic | +| State class | Any `AggregateState` | Must also implement `MemoryAwareState` | +| Built-in commands | None | `StoreMemoryCommand`, `ForgetMemoryCommand` | +| Built-in events | None | `MemoryStoredEvent`, `MemoryRevokedEvent` | +| Memory system | N/A | Sliding-window memory (configurable capacity) | +| External events | Via `@EventBridgeHandler` | Listens to **all** partitions directly | +| GDPR/PII | Supported | Not supported (agent memory is not PII) | +| Annotation | `@AggregateInfo` | `@AgenticAggregateInfo` | +| Kubernetes CRD | `AggregateService` | `AgenticAggregate` (short name `aag`) | + +### When to use an AgenticAggregate + +Use an AgenticAggregate when: + +- You need a stateful AI agent that remembers facts across conversations or tool invocations. +- The agent must react to domain events from other aggregates (e.g. to trigger analyses). +- You want the agent's memory to be durable, event-sourced, and auditable. +- You need exactly-one-instance semantics (the agent is a logical singleton in your domain). + +Do **not** use an AgenticAggregate for: + +- Standard business aggregates that need horizontal scaling across many entities. +- Aggregates that require GDPR key generation or PII encryption. +- Process managers (use `ProcessManager` instead). + +### Annotation + +```java +@AgenticAggregateInfo( + value = "MyAssistant", // Bean name (also used as Kafka topic prefix) + stateClass = MyAssistantState.class, + description = "My AI assistant", // Optional — used in EventCatalog docs + maxMemories = 100 // Sliding-window capacity (default: 100) +) +public final class MyAssistant implements AgenticAggregate { + // ... command handlers, event handlers ... +} +``` + +### State class + +The state class must implement both `AggregateState` and `MemoryAwareState`: + +```java +@AggregateStateInfo(type = "MyAssistantState", version = 1) +public record MyAssistantState( + String agenticAggregateId, + List memories +) implements AggregateState, MemoryAwareState { + + @Override public String getAggregateId() { return agenticAggregateId(); } + + @Override public List getMemories() { return memories(); } + + @Override + public MyAssistantState withMemory(AgenticAggregateMemory m) { + var updated = new ArrayList<>(memories); + updated.add(m); + return new MyAssistantState(agenticAggregateId, updated); + } + + @Override + public MyAssistantState withoutMemory(String memoryId) { + return new MyAssistantState(agenticAggregateId, + memories.stream().filter(m -> !m.memoryId().equals(memoryId)).toList()); + } +} +``` + +### Built-in memory commands and events + +The framework automatically handles: + +| Command / Event | Type | Description | +|-----------------|------|-------------| +| `StoreMemoryCommand` | Command | Asks the aggregate to store a new memory entry | +| `ForgetMemoryCommand` | Command | Asks the aggregate to remove an existing memory entry | +| `MemoryStoredEvent` | DomainEvent | Emitted when a memory entry is stored | +| `MemoryRevokedEvent` | DomainEvent | Emitted when a memory entry is removed (either by `ForgetMemoryCommand` or sliding-window eviction) | + +You do **not** need to implement command handlers for these; the +`KafkaAgenticAggregateRuntime` registers them as built-ins. + +### Sliding-window memory + +When the number of stored memories exceeds `maxMemories`, the oldest entry is automatically evicted +and a `MemoryRevokedEvent` is emitted. This keeps memory consumption bounded while preserving the +most recent context. + +### Listening to external events + +AgenticAggregates can react to domain events from other aggregates using `@EventHandler`-annotated +methods (same as regular aggregates). Internally, the runtime subscribes to **all** partitions of +the external event topic, ensuring the singleton receives every event regardless of which partition +it was written to. + +--- + +## Runtime Architecture + +### Kafka topics + +For each Aggregate `Foo` with `N` partitions the runtime creates: + +| Topic | Description | +|-------|-------------| +| `Foo-Commands` | Inbound command topic (N partitions) | +| `Foo-DomainEvents` | Event store topic (N partitions) | +| `Foo-AggregateState` | Compacted state snapshot topic (N partitions) | + +AgenticAggregates always use `N = 1`. + +### State storage + +Aggregate state is stored in **RocksDB** (embedded key–value store). State snapshots are published +to the `*-AggregateState` compacted topic so that a restarting instance can restore state without +replaying the entire event log. + +### Schema Registry + +All commands and events are serialised as **Avro** and registered in Confluent Schema Registry. +The `akces-eventcatalog` annotation processor additionally generates **JSON Schema** for +EventCatalog documentation at compile time. + +--- + +## EventCatalog Documentation + +The `akces-eventcatalog` annotation processor (`@AutoService(Processor.class)`) runs at +compile time and generates MDX documentation in `META-INF/eventcatalog/services//`. + +Supported annotations: + +| Annotation | Generated output | +|------------|-----------------| +| `@AggregateInfo` | `services//index.mdx` — standard aggregate service page | +| `@AgenticAggregateInfo` | `services//index.mdx` — singleton service page with `type: Singleton` and built-in memory commands/events pre-populated | +| `@CommandInfo` | `services//commands//index.mdx` + `schema.json` | +| `@DomainEventInfo` | `services//events//index.mdx` + `schema.json` | + +Configure the processor via `javac` annotation processor options: + +| Option | Default | Description | +|--------|---------|-------------| +| `eventcatalog.owners` | `framework-developers` | Comma-separated list of owners | +| `eventcatalog.repositoryBaseUrl` | GitHub URL | Base URL for source links | +| `eventcatalog.schemaDomain` | `akces-framework` | Schema domain prefix | diff --git a/README.md b/README.md index dce87419..32d875df 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ Akces leverages Kafka's distributed architecture for reliable event storage and - **Query Models**: Read-optimized projections of aggregate state - **Database Models**: Persistent storage of aggregate data optimized for queries - **Process Managers**: Coordinate workflows across multiple aggregates +- **AgenticAggregates**: Singleton aggregates with a built-in sliding-window memory system, designed for AI agent workflows (see [FRAMEWORK_OVERVIEW.md](FRAMEWORK_OVERVIEW.md)) ## Key Features @@ -84,6 +85,9 @@ Akces is organized into several Maven modules: - **client**: Client library for sending commands and processing responses - **query-support**: Support for query models and database models - **eventcatalog**: Annotation processor for generating API documentation +- **agentic**: AgenticAggregate support — singleton aggregates with AI-agent memory and MCP integration + +For a full architectural overview including AgenticAggregates, see [FRAMEWORK_OVERVIEW.md](FRAMEWORK_OVERVIEW.md). ## Getting Started diff --git a/SERVICES.md b/SERVICES.md new file mode 100644 index 00000000..7d61af3c --- /dev/null +++ b/SERVICES.md @@ -0,0 +1,259 @@ +# Akces Framework — Kubernetes Services + +This document describes the Kubernetes Custom Resource Definitions (CRDs) provided by the Akces +Operator and explains how to deploy each service type. For a general framework overview see +[FRAMEWORK_OVERVIEW.md](FRAMEWORK_OVERVIEW.md). + +--- + +## API Group + +All Akces CRDs are registered under: + +``` +apiVersion: akces.elasticsoftwarefoundation.org/v1 +``` + +--- + +## CRD Overview + +| Kind | Short name | Purpose | Replicas | +|------|-----------|---------|---------| +| `Aggregate` | `agg` | Standard event-sourced aggregate service | Configurable | +| `CommandService` | `cs` | Command validation and routing service | Configurable | +| `QueryService` | `qs` | Query model / read-model service | Configurable | +| `AgenticAggregate` | `aag` | Singleton AI-agent aggregate with memory | Always **1** | + +--- + +## Aggregate (`agg`) + +Deploys an event-sourced Aggregate service that processes commands and maintains aggregate state +via RocksDB snapshots. + +### Spec fields + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `image` | string | ✅ | Container image for the Aggregate service | +| `applicationName` | string | ✅ | Value for the `SPRING_APPLICATION_NAME` env var | +| `replicas` | integer | ✅ | Number of Kafka partitions / pod replicas | +| `args` | string[] | ❌ | Additional JVM / application arguments | +| `resources` | ResourceRequirements | ❌ | CPU and memory requests/limits | +| `enableSchemaOverwrites` | boolean | ❌ | Allow overwriting existing Avro schemas (default: `false`) | +| `env` | EnvVar[] | ❌ | Additional environment variables | +| `applicationProperties` | string | ❌ | Extra `application.properties` content merged into ConfigMap | +| `imagePullSecret` | string | ❌ | Name of the image pull secret | +| `storageClassName` | string | ❌ | StorageClass for the PVC (default: cluster default) | +| `storageSize` | string | ❌ | PVC size (default: `"4Gi"`) | + +### Example + +```yaml +apiVersion: akces.elasticsoftwarefoundation.org/v1 +kind: Aggregate +metadata: + name: wallet-aggregate + namespace: default +spec: + replicas: 3 + image: ghcr.io/elasticsoftwarefoundation/akces-aggregate-service:0.12.0 + applicationName: "Wallet Aggregate Service" + resources: + requests: + cpu: "500m" + memory: "512Mi" + limits: + cpu: "2" + memory: "2Gi" + storageSize: "8Gi" +``` + +--- + +## CommandService (`cs`) + +Deploys a command-routing service that validates incoming commands against their JSON Schema and +forwards them to the appropriate Aggregate topic. + +### Example + +```yaml +apiVersion: akces.elasticsoftwarefoundation.org/v1 +kind: CommandService +metadata: + name: wallet-commands + namespace: default +spec: + replicas: 2 + image: ghcr.io/elasticsoftwarefoundation/akces-command-service:0.12.0 + applicationName: "Wallet Command Service" +``` + +--- + +## QueryService (`qs`) + +Deploys a query-model service that consumes domain events and maintains read-optimised projections +in RocksDB or a relational database. + +### Example + +```yaml +apiVersion: akces.elasticsoftwarefoundation.org/v1 +kind: QueryService +metadata: + name: wallet-query + namespace: default +spec: + replicas: 1 + image: ghcr.io/elasticsoftwarefoundation/akces-query-service:0.12.0 + applicationName: "Wallet Query Service" +``` + +--- + +## AgenticAggregate (`aag`) + +Deploys a singleton AI-agent aggregate that has a built-in sliding-window memory system and can +communicate with external AI services (e.g. via the Model Context Protocol). See +[FRAMEWORK_OVERVIEW.md — AgenticAggregates](FRAMEWORK_OVERVIEW.md#agenticaggregates) for the +programming model. + +> **Note:** The replica count is always **1**. There is no `replicas` field in the spec — the +> operator always deploys exactly one pod backed by a single-partition StatefulSet. + +### Spec fields + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `image` | string | ✅ | Container image for the AgenticAggregate service | +| `applicationName` | string | ✅ | Value for the `SPRING_APPLICATION_NAME` env var | +| `args` | string[] | ❌ | Additional JVM / application arguments | +| `resources` | ResourceRequirements | ❌ | CPU and memory requests/limits for the main container | +| `enableSchemaOverwrites` | boolean | ❌ | Allow overwriting existing Avro schemas (default: `false`) | +| `sidecars` | SidecarSpec[] | ❌ | List of sidecar containers to inject (e.g. MCP server proxies) | +| `env` | EnvVar[] | ❌ | Additional environment variables for the main container | +| `applicationProperties` | string | ❌ | Extra `application.properties` content merged into ConfigMap | +| `imagePullSecret` | string | ❌ | Name of the image pull secret | +| `storageClassName` | string | ❌ | StorageClass for the PVC (default: cluster default) | +| `storageSize` | string | ❌ | PVC size (default: `"4Gi"`) | + +### SidecarSpec fields + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `name` | string | ✅ | Container name (must be unique within the pod) | +| `image` | string | ✅ | Container image for the sidecar | +| `env` | EnvVar[] | ❌ | Environment variables for the sidecar | +| `ports` | ContainerPort[] | ❌ | Ports to expose from the sidecar | +| `resources` | ResourceRequirements | ❌ | CPU and memory requests/limits for the sidecar | +| `readinessProbe` | Probe | ❌ | Kubernetes readiness probe | +| `livenessProbe` | Probe | ❌ | Kubernetes liveness probe | + +### Example — basic deployment + +```yaml +apiVersion: akces.elasticsoftwarefoundation.org/v1 +kind: AgenticAggregate +metadata: + name: my-assistant + namespace: default +spec: + image: ghcr.io/elasticsoftwarefoundation/akces-agentic-service:0.12.0 + applicationName: "My Assistant AgenticAggregate" + resources: + requests: + cpu: "500m" + memory: "512Mi" + limits: + cpu: "2" + memory: "2Gi" + applicationProperties: | + spring.ai.openai.api-key=${OPENAI_API_KEY} + spring.ai.openai.chat.options.model=gpt-4o + env: + - name: OPENAI_API_KEY + valueFrom: + secretKeyRef: + name: openai-credentials + key: api-key +``` + +### Example — with GitHub MCP Server sidecar + +The following example deploys an AgenticAggregate that uses the +[GitHub MCP Server](https://github.com/github/github-mcp-server) as a sidecar to give the AI +agent access to GitHub APIs via the Model Context Protocol (MCP). + +```yaml +apiVersion: akces.elasticsoftwarefoundation.org/v1 +kind: AgenticAggregate +metadata: + name: github-assistant + namespace: default +spec: + image: ghcr.io/elasticsoftwarefoundation/akces-agentic-service:0.12.0 + applicationName: "GitHub Assistant AgenticAggregate" + resources: + requests: + cpu: "500m" + memory: "512Mi" + limits: + cpu: "2" + memory: "2Gi" + sidecars: + - name: github-mcp-server + image: ghcr.io/github/github-mcp-server:latest + env: + - name: GITHUB_PERSONAL_ACCESS_TOKEN + valueFrom: + secretKeyRef: + name: github-credentials + key: token + ports: + - name: mcp + containerPort: 8080 + protocol: TCP + resources: + requests: + cpu: "100m" + memory: "128Mi" + limits: + cpu: "500m" + memory: "256Mi" + readinessProbe: + httpGet: + path: /health + port: 8080 + initialDelaySeconds: 5 + periodSeconds: 10 + applicationProperties: | + spring.ai.openai.api-key=${OPENAI_API_KEY} + spring.ai.openai.chat.options.model=gpt-4o + # Connect to the GitHub MCP Server sidecar running at localhost + spring.ai.mcp.client.sse.connections.github-mcp.url=http://localhost:8080 + env: + - name: OPENAI_API_KEY + valueFrom: + secretKeyRef: + name: openai-credentials + key: api-key +``` + +--- + +## Operator Deployment + +The Akces Operator itself is a standard Spring Boot application that watches the above CRDs and +reconciles the desired state (StatefulSet, Service, ConfigMap) in the cluster. + +```bash +# Install the operator using Helm (example — adapt to your release) +helm install akces-operator oci://ghcr.io/elasticsoftwarefoundation/akces-operator \ + --namespace akces-system --create-namespace \ + --set image.tag=0.12.0 +``` + +After the operator is running, apply your Custom Resources with `kubectl apply -f .yaml`. diff --git a/bom/pom.xml b/bom/pom.xml index 786500af..ef60c08b 100644 --- a/bom/pom.xml +++ b/bom/pom.xml @@ -61,6 +61,11 @@ akces-eventcatalog ${project.version} + + org.elasticsoftwarefoundation.akces + akces-agentic + ${project.version} + diff --git a/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessor.java b/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessor.java index 60e67f7b..3598891b 100644 --- a/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessor.java +++ b/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/EventCatalogProcessor.java @@ -39,6 +39,7 @@ "org.elasticsoftware.akces.annotations.CommandInfo", "org.elasticsoftware.akces.annotations.DomainEventInfo", "org.elasticsoftware.akces.annotations.AggregateInfo", + "org.elasticsoftware.akces.annotations.AgenticAggregateInfo", "org.elasticsoftware.akces.annotations.CommandHandler", "org.elasticsoftware.akces.annotations.EventHandler" }) @@ -54,6 +55,7 @@ public class EventCatalogProcessor extends AbstractProcessor { private final Map commandCache = new HashMap<>(); private final Map eventCache = new HashMap<>(); private final Map aggregateCache = new HashMap<>(); + private final Map agenticAggregateCache = new HashMap<>(); private final Map commandHandlerCache = new HashMap<>(); private final Map eventHandlerCache = new HashMap<>(); // Maps to store the relationships between aggregates and handlers @@ -64,6 +66,11 @@ public class EventCatalogProcessor extends AbstractProcessor { Map> aggregateErrorEvents = new HashMap<>(); Map> aggregateHandledCommands = new HashMap<>(); Map> aggregateHandledEvents = new HashMap<>(); + // Maps to store produced events and error events for each agentic aggregate + Map> agenticProducedEvents = new HashMap<>(); + Map> agenticErrorEvents = new HashMap<>(); + Map> agenticHandledCommands = new HashMap<>(); + Map> agenticHandledEvents = new HashMap<>(); // Create reverse lookup maps to find annotation details Map commandAnnotations = new HashMap<>(); Map eventAnnotations = new HashMap<>(); @@ -84,6 +91,8 @@ public boolean process(Set annotations, RoundEnvironment processDomainEventInfoAnnotations(roundEnv); } else if (AggregateInfo.class.getCanonicalName().equals(annotationName)) { processAggregateInfoAnnotations(roundEnv); + } else if (AgenticAggregateInfo.class.getCanonicalName().equals(annotationName)) { + processAgenticAggregateInfoAnnotations(roundEnv); } else if (CommandHandler.class.getCanonicalName().equals(annotationName)) { processCommandHandlerAnnotations(roundEnv); } else if (EventHandler.class.getCanonicalName().equals(annotationName)) { @@ -180,6 +189,32 @@ private void processAggregateInfoAnnotations(RoundEnvironment roundEnv) { } } + /** + * Scans all classes annotated with {@link AgenticAggregateInfo} and adds them to the + * {@code agenticAggregateCache} for subsequent documentation generation. + * + * @param roundEnv the annotation processing environment for the current round + */ + private void processAgenticAggregateInfoAnnotations(RoundEnvironment roundEnv) { + Set agenticElements = roundEnv.getElementsAnnotatedWith( + processingEnv.getElementUtils().getTypeElement(AgenticAggregateInfo.class.getCanonicalName()) + ); + + for (Element element : agenticElements) { + if (element.getKind() == ElementKind.CLASS) { + TypeElement typeElement = (TypeElement) element; + AgenticAggregateInfo agenticInfo = typeElement.getAnnotation(AgenticAggregateInfo.class); + String id = agenticInfo.value().isEmpty() ? typeElement.getSimpleName().toString() : agenticInfo.value(); + agenticAggregateCache.put(agenticInfo, typeElement); + + processingEnv.getMessager().printMessage( + Diagnostic.Kind.NOTE, + "Found agentic aggregate: " + id + " at " + typeElement.getQualifiedName() + ); + } + } + } + private void processCommandHandlerAnnotations(RoundEnvironment roundEnv) { Set handlerElements = roundEnv.getElementsAnnotatedWith( processingEnv.getElementUtils().getTypeElement(CommandHandler.class.getCanonicalName()) @@ -225,6 +260,12 @@ private void matchHandlersToAggregates() { typeToAggregate.put(entry.getValue(), entry.getKey()); } + // Create a map of TypeElement to AgenticAggregateInfo for quick lookup + Map typeToAgenticAggregate = new HashMap<>(); + for (Map.Entry entry : agenticAggregateCache.entrySet()) { + typeToAgenticAggregate.put(entry.getValue(), entry.getKey()); + } + // Match command handlers to aggregates for (Map.Entry entry : commandHandlerCache.entrySet()) { CommandHandler handler = entry.getKey(); @@ -246,6 +287,15 @@ private void matchHandlersToAggregates() { "Matched command handler " + methodElement + " to aggregate " + enclosingType.getQualifiedName() ); + } else { + AgenticAggregateInfo agenticInfo = typeToAgenticAggregate.get(enclosingType); + if (agenticInfo != null) { + processingEnv.getMessager().printMessage( + Diagnostic.Kind.NOTE, + "Matched command handler " + methodElement + " to agentic aggregate " + + enclosingType.getQualifiedName() + ); + } } } } @@ -271,6 +321,15 @@ private void matchHandlersToAggregates() { "Matched event handler " + methodElement + " to aggregate " + enclosingType.getQualifiedName() ); + } else { + AgenticAggregateInfo agenticInfo = typeToAgenticAggregate.get(enclosingType); + if (agenticInfo != null) { + processingEnv.getMessager().printMessage( + Diagnostic.Kind.NOTE, + "Matched event handler " + methodElement + " to agentic aggregate " + + enclosingType.getQualifiedName() + ); + } } } } @@ -603,6 +662,68 @@ private void generateCatalogEntries(String repositoryBaseUrl, List owner writeToEventCatalog(servicePath, "index.mdx", serviceDoc); } + // For each agentic aggregate, create service documentation marked as Singleton + for (Map.Entry entry : agenticAggregateCache.entrySet()) { + AgenticAggregateInfo agenticInfo = entry.getKey(); + TypeElement aggregateType = entry.getValue(); + + String aggregateName = agenticInfo.value().isEmpty() ? + aggregateType.getSimpleName().toString() : agenticInfo.value(); + + // Get the AgenticAggregateInfo annotation mirror to read stateClass. + AnnotationMirror agenticInfoMirror = getAnnotationMirror(aggregateType, AgenticAggregateInfo.class.getCanonicalName()); + TypeElement stateClass = getClassFromAnnotation(requireNonNull(agenticInfoMirror), "stateClass"); + AggregateStateInfo stateInfo = requireNonNull(stateClass).getAnnotation(AggregateStateInfo.class); + + // Built-in memory commands that every agentic aggregate receives + List builtInReceives = List.of( + new ServiceTemplateGenerator.Message("StoreMemory", "1.0.0"), + new ServiceTemplateGenerator.Message("ForgetMemory", "1.0.0") + ); + + // Built-in memory events that every agentic aggregate produces + List builtInSends = List.of( + new ServiceTemplateGenerator.Message("MemoryStored", "1.0.0"), + new ServiceTemplateGenerator.Message("MemoryRevoked", "1.0.0") + ); + + String servicePath = "META-INF/eventcatalog/services/" + aggregateName; + String serviceDoc = ServiceTemplateGenerator.generate(new ServiceTemplateGenerator.ServiceMetadata( + aggregateName, + stateInfo.version() + ".0.0", + aggregateName, + agenticInfo.description().isBlank() ? aggregateName + " AgenticAggregate" : agenticInfo.description(), + "Singleton", + owners, + Stream.concat( + builtInReceives.stream(), + agenticHandledCommands.getOrDefault(agenticInfo, Collections.emptySet()).stream() + .map(typeElement -> typeElement.getAnnotation(CommandInfo.class)) + .filter(Objects::nonNull) + .map(annotation -> new ServiceTemplateGenerator.Message( + annotation.type(), + annotation.version() + ".0.0")) + ).toList(), + Stream.concat( + builtInSends.stream(), + agenticProducedEvents.getOrDefault(agenticInfo, Collections.emptySet()).stream() + .map(typeElement -> typeElement.getAnnotation(DomainEventInfo.class)) + .filter(Objects::nonNull) + .map(annotation -> new ServiceTemplateGenerator.Message( + annotation.type(), + annotation.version() + ".0.0")) + ).toList(), + "Java", + repositoryBaseUrl + aggregateType.getQualifiedName().toString().replace('.', '/') + ".java")); + + processingEnv.getMessager().printMessage( + Diagnostic.Kind.NOTE, + "Generated service documentation for agentic aggregate " + aggregateName + ); + + writeToEventCatalog(servicePath, "index.mdx", serviceDoc); + } + // Generate command documentation and schemas for (Map.Entry entry : commandCache.entrySet()) { CommandInfo commandInfo = entry.getKey(); diff --git a/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/ServiceTemplateGenerator.java b/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/ServiceTemplateGenerator.java index e0c65c74..1dcda96d 100644 --- a/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/ServiceTemplateGenerator.java +++ b/main/eventcatalog/src/main/java/org/elasticsoftware/akces/eventcatalog/ServiceTemplateGenerator.java @@ -19,8 +19,20 @@ import java.util.List; +/** + * Generates EventCatalog-compatible MDX service documentation from service metadata. + *

+ * The generated output uses the EventCatalog MDX frontmatter format and includes + * an architecture node graph for visual representation. + */ public class ServiceTemplateGenerator { + /** + * Generates an MDX service documentation string from the given metadata. + * + * @param service the metadata describing the service (aggregate) + * @return the MDX-formatted service documentation string + */ public static String generate(ServiceMetadata service) { // Format the owners list StringBuilder ownersBuilder = new StringBuilder(); @@ -53,13 +65,18 @@ public static String generate(ServiceMetadata service) { } String sendsList = sendsBuilder.toString().stripTrailing(); - // Create the template with string replacements + // Build optional type line for the frontmatter + String typeLine = (service.type() != null && !service.type().isBlank()) + ? "type: " + service.type() + "\n" + : ""; + // Create the template with string replacements return SERVICE_TEMPLATE .replace("#{service.id}", service.id()) .replace("#{service.version}", service.version()) .replace("#{service.name}", service.name()) .replace("#{service.summary}", service.summary()) + .replace("#{typeLine}", typeLine) .replace("#{ownersList}", ownersList) .replace("#{receivesList}", receivesList) .replace("#{sendsList}", sendsList) @@ -67,7 +84,7 @@ public static String generate(ServiceMetadata service) { .replace("#{service.repositoryUrl}", service.repositoryUrl()); } - // Define POJO classes for the data model + /** Represents a command or event message entry in the catalog. */ public record Message(String id, String version) { } @@ -77,7 +94,7 @@ public record Message(String id, String version) { version: #{service.version} name: #{service.name} summary: #{service.summary} -owners: +#{typeLine}owners: #{ownersList} receives: #{receivesList} @@ -95,15 +112,57 @@ public record Message(String id, String version) {