Embabel Chat Store

Conversation persistence for the Embabel Agent framework: chat sessions, their messages, who sent each one to whom, attachments, structured data and assets, keyset paging, and a vector per message. It runs on five databases through Drivine4j, and in memory for development and tests.

Supported databases

Database	Notes
Neo4j	The default for this repository's tests. The message vector index needs 5.13 or later.
FalkorDB	Fastest on most workloads in graph-bench.
Memgraph	Use the memgraph/memgraph-mage image, which has the vector search module.
Oracle	Oracle Database 23ai or later. No Cypher: Drivine reads with SQL/PGQ and writes rows. Needs the JDBC driver. See Oracle.
Postgres	Postgres 14 or later. No Cypher: Drivine reads and writes with plain SQL, and searches vectors with pgvector. Needs the JDBC driver. See Postgres.

See it running

Guide is a working reference: an Embabel agent that answers questions about Embabel from its documentation, and keeps its conversations in this store on any of Neo4j, FalkorDB, Memgraph, Oracle or Postgres. Embabel Hub is its official front end.

Usage

Add the starter:

<dependency>
    <groupId>com.embabel.agent</groupId>
    <artifactId>embabel-agent-starter-chat-store</artifactId>
    <version>0.7.0-SNAPSHOT</version>
</dependency>

Configure a Drivine datasource:

database:
  datasources:
    neo:
      type: NEO4J          # or FALKORDB, MEMGRAPH; for ORACLE and POSTGRES see below
      host: localhost
      port: 7687
      user-name: neo4j
      password: secret
      database-name: neo4j

Define the repository. The auto-configuration builds the conversation factories on it:

@Bean
fun chatSessionRepository(
    factory: GraphObjectManagerFactory,
    persistenceManager: PersistenceManager,
    eventPublisher: ApplicationEventPublisher,
): ChatSessionRepository =
    ChatSessionRepositoryImpl(factory.stateless(), persistenceManager, eventPublisher)

Then create a conversation and add messages:

// Get a factory for your storage type
val factory = conversationFactoryProvider.get(ConversationStoreType.STORED)

// Create a conversation for a 1-1 chat
val conversation = (factory as StoredConversationFactory)
    .createForParticipants(
        id = sessionId,
        user = currentUser,
        agent = assistantUser,
        title = "My Chat"
    )

// Add messages - automatically routed based on role
conversation.addMessage(UserMessage("Hello!"))      // from=user, to=agent
conversation.addMessage(AssistantMessage("Hi!"))    // from=agent, to=user

The database is whichever one the datasource names. Nothing database-specific is passed in, so moving from one to another is a configuration change.

Oracle

Oracle Database 23ai or later. Oracle does not speak Cypher: Drivine reads it with SQL/PGQ and writes rows, and the repository is the same class as on the other databases.

1. Add the JDBC driver. Drivine does not bring it:

   <dependency>
       <groupId>com.oracle.database.jdbc</groupId>
       <artifactId>ojdbc11</artifactId>
   </dependency>

2. Point the datasource at the database:

   database:
     datasources:
       neo:
         type: ORACLE
         host: localhost
         port: 1521                  # the default
         database-name: FREEPDB1     # the service to connect to
         user-name: app
         password: secret

   Drivine makes the graph's tables and the property graph over them in this user's schema, the first time a manager speaks to the graph. The user needs the privileges to create them.

3. Define the repository as above.

Postgres

Postgres 14 or later. Postgres does not speak Cypher either: Drivine keeps the graph in tables of its own, reads and writes it with plain SQL, and the repository is the same class as on the other databases. The message vector index is an HNSW index of pgvector.

1. Add the JDBC driver. Drivine does not bring it:

   <dependency>
       <groupId>org.postgresql</groupId>
       <artifactId>postgresql</artifactId>
   </dependency>

2. Point the datasource at the database:

   database:
     datasources:
       neo:
         type: POSTGRES
         host: localhost
         port: 5432                  # the default
         database-name: app          # the database to connect to
         user-name: app
         password: secret

   Drivine makes the graph's tables the first time a manager speaks to the graph, and the pgvector extension where the server has it and the user may make it. Without pgvector everything works except the message vector index and vector search, and the index cannot be made, so turn it off with embabel.chat.store.vector-index.enabled: false or startup fails ensuring it.

3. Define the repository as above.

How it works

ConversationFactoryProvider
    ├── InMemoryConversationFactory  → InMemoryConversation  (ephemeral)
    └── StoredConversationFactory    → StoredConversation    (graph DB)
                                         │
                                         ├── ChatSessionRepository  (Drivine object manager)
                                         └── MessageEmbedder?       (optional: vector per message)

Type	Class	Use Case
IN_MEMORY	InMemoryConversation	Testing, ephemeral chats
STORED	StoredConversation	Production, persistent chats

A StoredConversation publishes each message as it is added, then persists it asynchronously through ChatSessionRepository: sessions, messages, paging, titles, narration and re-embedding.

Drivine

The model is a set of Drivine @NodeFragments (ChatSession, StoredMessage, User, StoredAsset, DataAttachment) and @GraphViews that load a node with what it is related to: a session with its owner and messages, a message with its author, recipients, attachments, data attachments and assets. ChatSessionRepositoryImpl reads and writes them through Drivine's StatelessGraphObjectManager and the query DSL Drivine generates at build time, so it holds no Cypher or SQL of its own. Drivine writes the statement for the configured database: Cypher for Neo4j, FalkorDB and Memgraph, SQL/PGQ for Oracle, plain SQL for Postgres.

The schema is declared, not written: the constraints, the lastActivityAt and kind range indexes and the message vector index are Drivine SchemaCatalogs, and Drivine's schema manager creates each in the database's own form at startup. A load or a save is one statement and one round trip however many relationships it carries.

Durable Assistant Assets

StoredConversation persists DurableAsset metadata carried by an AssistantMessage and restores those assets when the conversation is loaded again:

val message = AssistantMessage(
    content = "Here is the report",
    assets = listOf(durableAsset),
)
conversation.addMessage(message)

Asset metadata is stored in StoredAsset nodes connected to the message through HAS_ASSET. The content bytes are not stored in the graph. storageUri remains an opaque reference resolved by the AssetStore that originally materialized the asset.

Only DurableAsset instances are persisted. Ephemeral or application-specific Asset implementations remain available while the message is pending in memory but cannot be restored after a reload. Applications should materialize temporary assets before adding the assistant message to a stored conversation.

Deleting a session deletes its StoredAsset metadata nodes. It does not delete externally stored content; retention and content deletion remain the responsibility of the corresponding AssetStore.

Message Events

Subscribe to message lifecycle events for real-time updates:

@EventListener
fun onMessage(event: MessageEvent) {
    when (event.status) {
        MessageStatus.ADDED -> {
            // Message added - update UI immediately
            sendToWebSocket(event.toUserId, event)
        }
        MessageStatus.PERSISTED -> {
            // Message saved to storage
        }
        MessageStatus.PERSISTENCE_FAILED -> {
            // Handle error - event.error has details
        }
    }
}

Event Fields

Field	Description
conversationId	Session ID
fromUserId	Who sent the message
toUserId	Who should receive it (for routing). A message sent to several recipients names the first
title	Session title (for UI display)
message	The message content
status	ADDED, PERSISTED, or PERSISTENCE_FAILED

MessageEvent has room for one recipient. Once a message is persisted, a MessagePersistedEvent follows with its ID, its authorId and every one of its recipientIds, for listeners that route a multi-party message to each recipient.

Message Attribution

Messages have explicit sender and recipient:

(message:StoredMessage)-[:AUTHORED_BY]->(from:User)
(message:StoredMessage)-[:SENT_TO]->(to:User)          // one per recipient

Role-Based Routing (1-1 Chats)

Role	From	To
USER	user	agent
ASSISTANT	agent	user
SYSTEM	null	user

Multi-Party Chats

For multi-user or multi-agent scenarios, use addMessageFromTo for explicit routing:

// Create conversation without default participants
val conversation = factory.create(sessionId)

// Group chat with multiple users
conversation.addMessageFromTo(UserMessage("Hi everyone!"), from = alice, to = bob)
conversation.addMessageFromTo(UserMessage("Hello Alice!"), from = bob, to = alice)
conversation.addMessageFromTo(UserMessage("Hey both!"), from = charlie, to = alice)

// Agent handoff - one agent passing to another
conversation.addMessageFromTo(
    AssistantMessage("Transferring you to billing..."),
    from = supportAgent,
    to = customer
)
conversation.addMessageFromTo(
    AssistantMessage("Hi, I'm the billing specialist."),
    from = billingAgent,
    to = customer
)

// Multi-agent collaboration
conversation.addMessageFromTo(
    AssistantMessage("I'll research that."),
    from = researchAgent,
    to = coordinatorAgent
)
conversation.addMessageFromTo(
    AssistantMessage("Here's what I found..."),
    from = researchAgent,
    to = customer
)

Several recipients and structured data

addAttributedMessage names the author and every recipient of a message, and can attach structured data as well as files. It returns the message's ID:

val conversation = factory.create(sessionId) as StoredConversation

val record = DataAttachmentData.of(
    type = "entity",                      // the host's own vocabulary
    title = "Customer Northwind",
    json = """{"id":"c-1","risk":0.8}""",
)
val questionId = conversation.addAttributedMessage(
    UserMessage("@Triage @Steward is this customer at risk?"),
    from = alice,
    recipients = listOf(triageAgent, stewardAgent),
    dataAttachments = listOf(record),
)
conversation.addAttributedMessage(
    AssistantMessage("Yes, three invoices are overdue."),
    from = triageAgent,
    recipients = listOf(alice),
    inReplyTo = questionId,
)

conversation.entries.forEach { entry ->
    println("${entry.author?.displayName} -> ${entry.recipients.map { it.displayName }}: ${entry.message.content}")
}

A data attachment is stored in the graph with its message (HAS_DATA_ATTACHMENT), and is deleted with its session. Its type and json are opaque to the store. Make one per message: DataAttachmentData.of gives each a new ID, and an ID used on two messages is one node, deleted with either session. A message read back lists its recipients in no particular order.

ChatSessionRepository.addMessage(sessionId, MessageWrite(...)) and createSessionWithMessage(sessionId, owner, title, MessageWrite(...)) write the same at the repository level. A custom ChatSessionRepository that implements neither still takes a message with one recipient and no data attachment through them (and, for createSessionWithMessage, no file attachment), and refuses anything more with IllegalArgumentException.

Who took part

• ChatSessionRepository.getParticipants(sessionId) lists everyone who wrote in a session.
• SessionSummary.startedBy is the author of the earliest message that has one, read in the same query as the rest of the summary. A message with no author, such as a system greeting, is passed over; a session in which nobody has written yet has none.

Kinds of Session, and Sessions Within Sessions

A session's ID is an opaque UUIDv7 and says nothing about the session. What a session is lives in the graph: its kind, the session it is PART_OF, and the user it is held WITH.

flowchart LR
    owner(["User<br/>(the person)"])
    agent(["User<br/>(an agent)"])
    thread["ChatSession<br/>kind: thread"]
    side["ChatSession<br/>kind: chat"]
    msg["StoredMessage"]

    thread -- OWNED_BY --> owner
    side -- OWNED_BY --> owner
    side -- PART_OF --> thread
    side -- WITH --> agent
    thread -- HAS_MESSAGE --> msg
    msg -- AUTHORED_BY --> agent
    msg -- SENT_TO --> agent

Loading

Don't put meaning in a session ID, such as a th_ prefix for a thread: a prefixed ID is no longer time-ordered, so it breaks SessionOrder.CREATED, and what it says can't change without changing the ID. All three of kind, PART_OF and WITH are set when a session is created and never change:

repository.createSession(threadId, owner, "Planning", SessionWrite(kind = "thread"))

• kind is any string the app chooses, chat (SessionData.DEFAULT_KIND) unless given.
• PART_OF links a session to the one it belongs to, such as an agent's private conversation inside a thread. The session it is part of must exist, must itself be top-level, and must have the same owner.
• WITH links a session to the one user it is held with, such as that agent.

An agent's conversation inside a thread is found again by the thread and the agent, and made the first time:

val side = repository.findOrCreatePart(partOf = threadId, heldWith = agent)   // owned by the thread's owner
repository.findPart(threadId, agent.id)                                   // Optional, never creates
repository.listParts(threadId)                                            // every part, as summaries

No database can make the (thread, agent) pair unique, so two callers at once can each make one. Every lookup returns the earliest, so they settle on the same session.

Listings. listSessionsForUser, listSessionSummariesForUser and countSessionsForUser return top-level sessions only, so side conversations don't crowd the owner's list, and a count always matches its list. Pass a SessionFilter to choose a kind or to include the sessions that are part of another, as a parameter or as SessionPageRequest.filter:

repository.listSessionSummariesForUser(userId, SessionOrder.LAST_ACTIVITY, SessionFilter(kind = "thread"))
repository.countSessionsForUser(userId, SessionFilter(includeParts = true))

Deleting a session deletes the sessions that are part of it, with their messages. The users they are owned by, held with or written by are kept. Since a session is part of a top-level session only, the delete reaches one level; if something written below the repository ever nested a session deeper, the delete is refused before anything is written rather than leaving that session behind.

Upgrading. A session stored before kind existed reads as chat, and a filter for chat returns it, so nothing needs to run first. ChatSessionRepository.backfillSessionKinds() writes the value onto such sessions a page at a time; it is safe to run again, and returns how many it wrote.

A custom ChatSessionRepository keeps compiling: the new methods have defaults. It takes only SessionWrite.DEFAULT, finds no parts, and refuses findOrCreatePart with UnsupportedOperationException.

Sorting and Pagination

Two orderings

Session listings support two orderings, selected with SessionOrder:

SessionOrder	Keyset	Meaning
CREATED	sessionId	Newest-created first (default)
LAST_ACTIVITY	(lastActivityAt, sessionId)	Most recently active first

CREATED keys on the session ID alone. Session IDs are UUIDv7, which embeds a millisecond timestamp in the leading 48 bits, and the canonical string form is big-endian hex — so lexicographic comparison is chronological comparison. The ID is unique, so this is already a total order and needs no tie-breaker, and it is served by the index backing the sessionId uniqueness constraint. Two sessions created in the same millisecond order by the random bits rather than by true creation time: arbitrary, but stable.

LAST_ACTIVITY orders by lastActivityAt descending, with sessionId descending purely as a tie-breaker — the ID is not the primary sort key here.

Message order within a thread is always by messageId and is not configurable. Since message IDs are also UUIDv7, a thread always reads in the order it happened.

What counts as activity

lastActivityAt advances only when a message is added. It is deliberately not touched by:

• narration (updateMessageNarration)
• title generation or renaming
• any other enrichment of an existing message

Editing or annotating a message should not jump its session to the top of a list, any more than it should move the message within its thread. The timestamp is also monotonic — it only ever moves forward — so clock skew between application instances can leave ordering slightly stale but can never move a session backwards, which would let a keyset walk skip or repeat it.

Paging

Paging is keyset-based, not offset-based, so page n costs the same as page 1:

var cursor: String? = null
do {
    val page = repository.listSessionsForUser(
        userId,
        SessionPageRequest(pageSize = 20, cursor = cursor, order = SessionOrder.CREATED),
    )
    page.items.forEach { render(it) }
    cursor = page.nextCursor
} while (cursor != null)

nextCursor is null on the last page. Use listSessionSummariesForUser for the same paging over lightweight summaries — owner, a message count and who started the session, computed in the query, without loading message bodies.

Cursors are opaque: a versioned, Base64URL-encoded binary payload. Do not parse or construct them. Each cursor also records the SessionOrder that issued it, and is rejected if replayed against the other ordering — the two key on different properties, so silently seeking on the wrong one would return a plausible but wrong page.

There is no snapshot isolation across requests. Under LAST_ACTIVITY, a session that becomes active mid-walk can move ahead of a cursor already issued and so be seen twice or missed. CREATED keys on an immutable value and does not have this property.

Upgrading from before lastActivityAt

SessionData.lastActivityAt carries Drivine's @Default, so a session stored before the property existed hydrates as its createdAt instead of failing to load. Old data therefore stays readable with no migration, and sorts sensibly under CREATED — which is another reason that is the default ordering.

LAST_ACTIVITY needs more: a keyset comparison is never satisfied by a value that is absent in the graph, so a session whose property is only defaulted client-side drops out of every page after the first. Every session written since 0.5.0 has the property, and an older one gains it when it next receives a message.

Releases 0.5.0 and 0.6.0 backfilled the property at startup, for sessions stored before it existed. That backfill ran once for a database and has been removed: a database that still holds sessions from before 0.5.0 should be started once on 0.6.0 before it is upgraded further.

Concurrent Writes

SessionData and MessageData each have a stamp field, annotated with Drivine's @NodeStamp. Loading fills it. A save of a session or message that carries a stamp is applied only if the node's properties are as they were when it was loaded, and is otherwise refused with StaleObjectException. Nothing is written by a refused save.

val session = repository.findBySessionId(sessionId).get().session
// ... another writer adds a message, which advances lastActivityAt ...
graphObjectManager.save(session.copy(title = "New title"))   // StaleObjectException

The repository's own writes need no handling:

Write	What it does about a concurrent change
updateSessionTitle	Writes only the title. If the session changed after it was loaded, it is loaded again and the title applied to the fresh one.
addMessage	Adds the message, and advances lastActivityAt on the session node alone, in one statement that compares with the stored value, so it only moves forward.
updateMessageNarration	Writes only the narration of the one message, in one statement, and only if the message is in the session named.
reembedMessages	Writes only the vector and its model, in one statement per message, so a message changed after its page was loaded keeps the change.

Treat the stamp as opaque, and leave it null on a session or message you construct. It is part of equals, copy, and any JSON made from these classes directly.

Cypher of your own

A save notices only a write that replaced the stamp. Cypher that sets a property of a ChatSession or StoredMessage node must therefore set Stamps.setClause in the same statement:

"MATCH (m:StoredMessage {messageId: \$id}) SET m.narration = \$narration, ${Stamps.setClause("m")}"

Without it, a save of an object loaded before the statement ran overwrites what it wrote.

The repository itself writes no Cypher: each of its reads and writes goes through Drivine's object manager, which marks the stamp and runs on every store Drivine supports.

Upgrading from before stamps

ChatSessionRepositoryImpl takes a StatelessGraphObjectManager, where it took a GraphObjectManager:

@Bean
fun chatSessionRepository(
    factory: GraphObjectManagerFactory,
    persistenceManager: PersistenceManager,
): ChatSessionRepository = ChatSessionRepositoryImpl(factory.stateless(), persistenceManager)

Stored data needs no migration. A session or message stored before stamps loads with a null stamp, so its first save is not checked, and that save gives the node a stamp.

Auto Title Generation

Provide a TitleGenerator to automatically generate session titles:

val factory = StoredConversationFactory(
    repository = chatSessionRepository,
    eventPublisher = applicationEventPublisher,
    titleGenerator = { message ->
        // Generate title from first message
        llm.generate("Summarize in 5 words: ${message.content}")
    }
)

Vector Embeddings on Messages

When a MessageEmbedder is wired, every persisted message is embedded inline within the existing async-persistence coroutine — the embedding lands with the message in a single DB write, no extra round-trip per message.

addMessage()
  ├─ MessageEvent(ADDED)              ← synchronous (front-ends update here)
  └─ async {
       embedder.embed(message)        ← inline, before the DB write
       repository.addMessage(... embedding ...)
       MessageEvent(PERSISTED)
     }

The embedding is stored as two properties on the :StoredMessage node:

(msg:StoredMessage {
  messageId, role, content, createdAt,
  embedding,                          // List<Float>
  embeddingModel                      // EmbeddingService.name, for drift detection
})

embedding and embeddingModel are both nullable. Messages without an embedding (failure, no embedder configured, SYSTEM/blank content) just have null values.

When embedding a message fails, the message is still saved, with a null embeddingModel, a WARN is logged with its id, and a MessageEmbeddingFailedEvent (session id, message id, role, cause) is published once the write lands. A null model is what reembedMessages selects on, so the retry is a re-embed run with the current model — MessageReembedder.reembed(embeddingService) — triggered from a listener for that event or on a schedule. A run only embeds what is pending.

reembedMessages splits a failing embed call down to single messages, writes everything that embeds, always remakes the message index, and then throws MessageReembedIncompleteException naming the messages that failed (and, if it gave up on a dead service, how many it never tried).

Embedder Configuration

The auto-configuration wires a default embedder when an embabel Ai bean is available:

RoleFilteringMessageEmbedder(           // Skips SYSTEM and blank-content messages
    DefaultMessageEmbedder(             // Calls ai.withDefaultEmbeddingService()
        ai.withDefaultEmbeddingService()
    )
)

To override — embed all roles, use a different embedding service, etc. — define your own MessageEmbedder bean and the autoconfig backs off.

Failure Mode

Embedding failures are caught and logged; the message is persisted with a null embedding. Messages are never lost because an embedding call failed.

Vector Index Management

The chat store declares a vector index on :StoredMessage(embedding) as a Drivine SchemaCatalog, and Drivine's schema manager ensures it at startup, in the form the configured database takes.

embabel:
  chat:
    store:
      vector-index:
        enabled: true                   # default
        label: StoredMessage            # default
        property: embedding             # default
        similarity-function: cosine     # cosine | euclidean
        name: null                      # default: Drivine derives one from label and property

The index is sized to the embedding model's dimensions and versioned by the model's name. With no embedding model configured, no index is declared, since an index at the wrong width would accept writes. Ensuring is idempotent: a matching index is left alone, and an index of a different shape is reported as drift, not dropped.

Changing the embedding model

Re-embed the stored messages with the new model:

messageReembedder.reembed(embeddingService)

MessageReembedder remakes the vector index from the same spec the startup catalog uses, so the two cannot describe different indexes.

User Implementation

Implement StoredUser for your user type. The StoredUser interface extends User from embabel-agent-api with Drivine annotations for graph-database persistence (the same annotations work on every supported database):

@NodeFragment(labels = ["User", "MyUser"])
data class MyUser(
    @NodeId override val id: String,
    override val displayName: String,
    override val username: String,
    override val email: String?
) : StoredUser

Register it with Drivine, so a User node with the MyUser label loads as a MyUser:

persistenceManager.registerSubtype(
    StoredUser::class.java,
    listOf("User", "MyUser"),   // the subtype's labels, in any order
    MyUser::class.java,
)

Drivine registers subtypes itself only for a sealed class or a base type annotated with @JsonSubTypes. StoredUser is an interface applications implement, so it is neither, and an implementation needs this call.

For simple cases without extra fields, use the built-in SimpleStoredUser:

val user = SimpleStoredUser(
    id = "user-123",
    displayName = "Alice",
    username = "alice",
    email = "alice@example.com"
)

Spring Boot Auto-Configuration

Add the dependency and configure:

embabel:
  chat:
    store:
      enabled: true                  # default
      title-after-message-count: 1   # default: regenerate title every N messages
      vector-index:
        enabled: true                # default: declare the message vector index
        similarity-function: cosine  # default

Beans auto-configured:

Bean	Conditional on	Purpose
storedConversationFactory	ChatSessionRepository	Creates persistent conversations
inMemoryConversationFactory	—	Creates ephemeral conversations
conversationFactoryProvider	—	Aggregates all factories
titleGenerator	Ai bean	LLM-driven title generation
messageEmbedder	Ai bean	Inline embedding on persisted messages
chatStoreConstraintSchema	—	Uniqueness constraints and the lastActivityAt range index
chatStoreVectorIndexSchema	Ai bean + property	The message vector index, at the embedding model's width
messageReembedder	ChatSessionRepository	Re-embeds messages and remakes the vector index
sessionEventAwaiter	—	Waits on a session's persistence events

Each of these can be overridden by defining your own bean of the same type.

Testing

Integration tests use Testcontainers, on Neo4j, Oracle Free and Postgres with pgvector:

mvn test

The Oracle image is large and the database wants two gigabytes. To run one store's tests alone:

mvn -pl embabel-chat-store test -Dtest='*OracleTest'      # or '*PostgresTest'

Dependencies

• embabel-agent-api - Core conversation interfaces
• drivine - graph persistence on Neo4j, FalkorDB, Memgraph, Oracle and Postgres
• Spring Boot (optional, for auto-configuration)