Persist durable assistant assets

README.md

@@ -44,6 +44,31 @@ ConversationFactoryProvider
 | `IN_MEMORY` | `InMemoryConversation` | Testing, ephemeral chats |
 | `STORED` | `StoredConversation` | Production, persistent chats |
 
+## Durable Assistant Assets
+
+`StoredConversation` persists `DurableAsset` metadata carried by an `AssistantMessage` and
+restores those assets when the conversation is loaded again:
+
+```kotlin
+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:

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/adapter/StoredConversation.kt

@@ -16,15 +16,18 @@
 package com.embabel.chat.store.adapter
 
 import com.embabel.agent.api.identity.User
+import com.embabel.chat.AssistantMessage
 import com.embabel.chat.AssetTracker
 import com.embabel.chat.Conversation
+import com.embabel.chat.DurableAsset
 import com.embabel.chat.Message
 import com.embabel.chat.MessageRole
 import com.embabel.chat.event.MessageEvent
 import com.embabel.chat.store.embedding.MessageEmbedder
 import com.embabel.chat.store.event.MessagePersistedEvent
 import com.embabel.chat.store.event.SessionEventAwaiter
 import com.embabel.chat.store.model.AttachmentData
+import com.embabel.chat.store.model.AssetData
 import com.embabel.chat.store.model.MessageData
 import com.embabel.chat.store.model.StoredSession
 import com.embabel.chat.store.model.StoredUser
@@ -101,7 +104,7 @@ class StoredConversation(
 
     private val logger = LoggerFactory.getLogger(StoredConversation::class.java)
 
-    private val pendingMessages = ConcurrentLinkedQueue<MessageData>()
+    private val pendingMessages = ConcurrentLinkedQueue<PendingMessage>()
 
     /**
      * Returns true since this conversation is backed by persistent storage.
@@ -116,8 +119,8 @@ class StoredConversation(
         get() {
             val dbMessages = repository.getMessages(id)
             val dbMessageIds = dbMessages.mapTo(HashSet()) { it.messageId }
-            val pending = pendingMessages.filter { it.messageId !in dbMessageIds }
-            return dbMessages.map { it.toMessage() } + pending.map { it.toMessage() }
+            val pending = pendingMessages.filter { it.messageData.messageId !in dbMessageIds }
+            return dbMessages.map { it.toMessage() } + pending.map { it.message }
         }
 
     /**
@@ -257,6 +260,17 @@ class StoredConversation(
         attachments: List<AttachmentData> = emptyList()
     ): String {
         val messageData = MessageData.from(message, messageId = UUIDv7.generateString())
+        val messageAssets = (message as? AssistantMessage)?.assets.orEmpty()
+        val durableAssets = messageAssets.filterIsInstance<DurableAsset>().map {
+            AssetData.from(it, messageData.messageId)
+        }
+        if (messageAssets.size != durableAssets.size) {
+            logger.warn(
+                "Message {} has {} non-durable assets that will not survive conversation reload",
+                messageData.messageId,
+                messageAssets.size - durableAssets.size,
+            )
+        }
 
         // Generate an interim title from the first user message so the session
         // appears immediately in the UI.  The LLM will replace it later once
@@ -281,17 +295,26 @@ class StoredConversation(
         )
 
         // Add to pending buffer so getMessages() returns this message immediately
-        pendingMessages.add(messageData)
+        val pendingMessage = PendingMessage(messageData, message)
+        pendingMessages.add(pendingMessage)
 
         // DB write — asynchronous, non-blocking. The pending buffer ensures
         // consistent reads while the write is in flight.
         scope.launch {
             try {
                 val messageDataForDb = embedMessageData(message, messageData)
-                val updatedSession = addMessageWithAwait(id, messageDataForDb, from, to, signal, attachments)
+                val updatedSession = addMessageWithAwait(
+                    id,
+                    messageDataForDb,
+                    from,
+                    to,
+                    signal,
+                    attachments,
+                    durableAssets,
+                )
 
                 // Persisted — remove from pending buffer (DB is now the source of truth)
-                pendingMessages.remove(messageData)
+                pendingMessages.remove(pendingMessage)
 
                 // PERSISTED event
                 try {
@@ -381,10 +404,11 @@ class StoredConversation(
         author: StoredUser?,
         recipient: StoredUser?,
         signal: CompletableDeferred<Unit>,
-        attachments: List<AttachmentData> = emptyList()
+        attachments: List<AttachmentData> = emptyList(),
+        assets: List<AssetData> = emptyList(),
     ): StoredSession {
         return try {
-            repository.addMessage(sessionId, messageData, author, recipient, attachments)
+            persistMessage(sessionId, messageData, author, recipient, attachments, assets)
         } catch (e: IllegalArgumentException) {
             if (e.message?.contains("Session not found") != true) throw e
 
@@ -393,10 +417,28 @@ class StoredConversation(
                 sessionId, messageData.messageId
             )
             sessionEventAwaiter.awaitSession(signal)
-            repository.addMessage(sessionId, messageData, author, recipient, attachments)
+            persistMessage(sessionId, messageData, author, recipient, attachments, assets)
         }
     }
 
+    private fun persistMessage(
+        sessionId: String,
+        messageData: MessageData,
+        author: StoredUser?,
+        recipient: StoredUser?,
+        attachments: List<AttachmentData>,
+        assets: List<AssetData>,
+    ): StoredSession = if (assets.isEmpty()) {
+        repository.addMessage(sessionId, messageData, author, recipient, attachments)
+    } else {
+        repository.addMessageWithAssets(sessionId, messageData, author, recipient, attachments, assets)
+    }
+
+    private data class PendingMessage(
+        val messageData: MessageData,
+        val message: Message,
+    )
+
     companion object {
         /**
          * Truncate user message content into a short interim title.

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/model/AssetData.kt

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2024-2026 Embabel Pty Ltd.
+ *
+ * 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 com.embabel.chat.store.model
+
+import com.embabel.chat.DurableAsset
+import org.drivine.annotation.NodeFragment
+import org.drivine.annotation.NodeId
+import java.time.Instant
+
+/**
+ * Persisted metadata for a [DurableAsset]. Content remains in the external
+ * [storageUri] managed by the originating asset store.
+ */
+@NodeFragment(labels = ["StoredAsset"])
+data class AssetData(
+    @NodeId val storedAssetId: String,
+    val assetId: String,
+    val name: String,
+    val mimeType: String?,
+    val sizeBytes: Long,
+    val contentHash: String,
+    val storageUri: String,
+    val createdAt: Instant,
+) {
+
+    fun toAsset(): DurableAsset = DurableAsset(
+        id = assetId,
+        name = name,
+        mimeType = mimeType,
+        sizeBytes = sizeBytes,
+        contentHash = contentHash,
+        storageUri = storageUri,
+        timestamp = createdAt,
+    )
+
+    companion object {
+
+        @JvmStatic
+        fun from(asset: DurableAsset, messageId: String): AssetData = AssetData(
+            storedAssetId = "$messageId:${asset.id}",
+            assetId = asset.id,
+            name = asset.name,
+            mimeType = asset.mimeType,
+            sizeBytes = asset.sizeBytes,
+            contentHash = asset.contentHash,
+            storageUri = asset.storageUri,
+            createdAt = asset.timestamp,
+        )
+    }
+}

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/model/DeletableMessage.kt

@@ -21,8 +21,8 @@ import org.drivine.annotation.GraphView
 import org.drivine.annotation.Root
 
 /**
- * Delete-only view of a message: the `:StoredMessage` node plus the `:Attachment` nodes
- * that hang off it, and nothing else.
+ * Delete-only view of a message: the `:StoredMessage` node plus its owned attachment
+ * and durable asset metadata nodes, and nothing else.
  *
  * Exists because the cascade boundary IS the view shape. [DeletableSession] previously
  * declared its children as bare [MessageData], which was correct while a message owned no
@@ -37,6 +37,7 @@ import org.drivine.annotation.Root
  * Neo4j structure traversed:
  * ```
  * (msg:StoredMessage)-[:HAS_ATTACHMENT]->(att:Attachment)
+ * (msg:StoredMessage)-[:HAS_ASSET]->(asset:StoredAsset)
  * ```
  *
  * @see DeletableSession for the session-level cascade this composes into
@@ -52,5 +53,12 @@ data class DeletableMessage(
      * The message's attachments, deleted with it.
      */
     @GraphRelationship(type = "HAS_ATTACHMENT", direction = Direction.OUTGOING)
-    val attachments: List<AttachmentData> = emptyList()
+    val attachments: List<AttachmentData> = emptyList(),
+
+    /**
+     * The message's durable asset metadata, deleted with it.
+     * External content retention remains the asset store's responsibility.
+     */
+    @GraphRelationship(type = "HAS_ASSET", direction = Direction.OUTGOING)
+    val assets: List<AssetData> = emptyList(),
 )

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/model/DeletableSession.kt

@@ -29,21 +29,20 @@ import org.drivine.annotation.Root
  *
  * Deliberately narrow:
  * - The child is [DeletableMessage], which carries the `:StoredMessage` node and its
- *   HAS_ATTACHMENT children — but **not** [SimpleStoredMessage] (which carries
+ *   HAS_ATTACHMENT and HAS_ASSET children — but **not** [SimpleStoredMessage] (which carries
  *   AUTHORED_BY/SENT_TO → [StoredUser]). A DETACH DELETE of each message node drops
  *   those edges while leaving the `:User` nodes intact.
  * - The OWNED_BY owner is omitted entirely. Because the view declares only
  *   HAS_MESSAGE, the cascade physically cannot reach `:User` — the safety is
  *   structural, not a flag.
  *
- * Attachments are included precisely because that structural narrowness cuts both ways:
- * a message's attachments are owned by the message and reachable from nowhere else, so a
- * cascade that stops at `:StoredMessage` leaves `:Attachment` nodes orphaned, referencing
- * stored bytes nothing can reach.
+ * Attachments and asset metadata are included precisely because that structural narrowness
+ * cuts both ways: they are owned by the message and reachable from nowhere else.
  *
  * Neo4j structure traversed:
  * ```
  * (session:ChatSession)-[:HAS_MESSAGE]->(msg:StoredMessage)-[:HAS_ATTACHMENT]->(att:Attachment)
+ *                                                    \-[:HAS_ASSET]->(asset:StoredAsset)
  * ```
  *
  * @see com.embabel.chat.store.model.StoredSession for the full read view

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/model/StoredMessage.kt

@@ -16,6 +16,7 @@
 package com.embabel.chat.store.model
 
 import com.embabel.chat.AssistantMessage
+import com.embabel.chat.DurableAsset
 import com.embabel.chat.Message
 import com.embabel.chat.MessageRole
 import com.embabel.chat.SystemMessage
@@ -110,9 +111,10 @@ data class MessageData(
     /**
      * Convert to a rich agent-api [Message] type.
      */
-    fun toMessage(): Message = when (role) {
+    @JvmOverloads
+    fun toMessage(assets: List<DurableAsset> = emptyList()): Message = when (role) {
         MessageRole.USER -> UserMessage(content = content, timestamp = createdAt)
-        MessageRole.ASSISTANT -> AssistantMessage(content = content, timestamp = createdAt)
+        MessageRole.ASSISTANT -> AssistantMessage(content = content, assets = assets, timestamp = createdAt)
         MessageRole.SYSTEM -> SystemMessage(content = content, timestamp = createdAt)
     }
 }
@@ -162,7 +164,11 @@ data class SimpleStoredMessage(
      * reload rather than existing only for the turn it arrived in.
      */
     @GraphRelationship(type = "HAS_ATTACHMENT", direction = Direction.OUTGOING)
-    val attachments: List<AttachmentData> = emptyList()
+    val attachments: List<AttachmentData> = emptyList(),
+
+    /** Durable assets produced with this assistant message. */
+    @GraphRelationship(type = "HAS_ASSET", direction = Direction.OUTGOING)
+    val assets: List<AssetData> = emptyList(),
 ) : Message {
     // Convenience accessors for common properties
     val messageId: String get() = message.messageId
@@ -177,7 +183,7 @@ data class SimpleStoredMessage(
     /**
      * Convert to a rich agent-api [Message] type.
      */
-    fun toMessage(): Message = message.toMessage()
+    fun toMessage(): Message = message.toMessage(assets.map { it.toAsset() })
 }
 
 /**
@@ -200,7 +206,11 @@ data class AttributedMessage(
 
     /** Files attached to this message. See [SimpleStoredMessage.attachments]. */
     @GraphRelationship(type = "HAS_ATTACHMENT", direction = Direction.OUTGOING)
-    val attachments: List<AttachmentData> = emptyList()
+    val attachments: List<AttachmentData> = emptyList(),
+
+    /** Durable assets produced with this assistant message. */
+    @GraphRelationship(type = "HAS_ASSET", direction = Direction.OUTGOING)
+    val assets: List<AssetData> = emptyList(),
 )
 
 /**

embabel-chat-store/src/main/kotlin/com/embabel/chat/store/repository/ChatSessionRepository.kt

@@ -16,6 +16,7 @@
 package com.embabel.chat.store.repository
 
 import com.embabel.chat.store.model.AttachmentData
+import com.embabel.chat.store.model.AssetData
 import com.embabel.chat.store.model.MessageData
 import com.embabel.chat.store.model.SessionSummary
 import com.embabel.chat.store.model.SimpleStoredMessage
@@ -165,6 +166,21 @@ interface ChatSessionRepository {
         attachments: List<AttachmentData> = emptyList()
     ): StoredSession
 
+    /**
+     * Add a message and its durable asset metadata to a session.
+     *
+     * This is separate from [addMessage] to preserve the existing message and attachment
+     * write path for callers that do not produce assets.
+     */
+    fun addMessageWithAssets(
+        sessionId: String,
+        messageData: MessageData,
+        author: StoredUser? = null,
+        recipient: StoredUser? = null,
+        attachments: List<AttachmentData> = emptyList(),
+        assets: List<AssetData>,
+    ): StoredSession
+
     /**
      * Get all messages in a session.
      *