fix(core): retain buffered messages until memory writes succeed

.changeset/memory-persist-retry.md

@@ -0,0 +1,5 @@
+---
+"@voltagent/core": patch
+---
+
+Keep conversation messages pending until their memory writes succeed and retry failed writes on the next operation for the same conversation. Memory failures remain best effort for agent responses.

packages/core/src/agent/agent.spec.ts

@@ -2532,6 +2532,67 @@ Use pandas and summarize findings.`.split("\n"),
       );
     });
 
+    it("returns the generated response after a memory write fails and saves it on the next turn", async () => {
+      const memory = new Memory({ storage: new InMemoryStorageAdapter() });
+      const saveMessage = memory.saveMessageWithContext.bind(memory);
+      let failAssistantWrite = true;
+      vi.spyOn(memory, "saveMessageWithContext").mockImplementation(async (...args) => {
+        if (failAssistantWrite && args[0].role === "assistant") {
+          failAssistantWrite = false;
+          throw new Error("storage unavailable");
+        }
+        return saveMessage(...args);
+      });
+      const agent = new Agent({
+        name: "TestAgent",
+        instructions: "Test",
+        model: mockModel as any,
+        memory,
+      });
+      const response = (text: string) => ({
+        text,
+        content: [{ type: "text", text }],
+        reasoning: [],
+        files: [],
+        sources: [],
+        toolCalls: [],
+        toolResults: [],
+        finishReason: "stop",
+        usage: providerUsage,
+        warnings: [],
+        request: {},
+        response: {
+          id: text,
+          modelId: "test-model",
+          timestamp: new Date(),
+          messages: createAssistantResponseMessages(text),
+        },
+        steps: [],
+      });
+      vi.mocked(ai.generateText)
+        .mockResolvedValueOnce(response("First response") as any)
+        .mockResolvedValueOnce(response("Second response") as any);
+      const options = {
+        memory: {
+          userId: "retry-user",
+          conversationId: "retry-conversation",
+          options: { messageMetadataPersistence: true },
+        },
+      };
+
+      const first = await agent.generateText("First turn", options);
+      expect(first.text).toBe("First response");
+
+      const second = await agent.generateText("Second turn", options);
+      expect(second.text).toBe("Second response");
+
+      const saved = await memory.getMessages("retry-user", "retry-conversation");
+      expect(saved.filter((message) => message.role === "assistant")).toEqual([
+        expect.objectContaining({ parts: [{ type: "text", text: "First response" }] }),
+        expect.objectContaining({ parts: [{ type: "text", text: "Second response" }] }),
+      ]);
+    });
+
     it("should persist usage and finish reason in assistant message metadata for streamText", async () => {
       const memory = new Memory({
         storage: new InMemoryStorageAdapter(),

packages/core/src/agent/agent.ts

@@ -1485,7 +1485,7 @@ export class Agent {
             void this.recordStepResults(result.steps, oc);
 
             if (!shouldDeferPersist && shouldPersistMemory) {
-              await persistQueue.flush(buffer, oc);
+              await this.flushConversationMessagesBestEffort(persistQueue, buffer, oc);
             }
 
             const finalText = await executeOutputGuardrails({
@@ -1585,7 +1585,7 @@ export class Agent {
             }
 
             if (shouldDeferPersist && shouldPersistMemory) {
-              await persistQueue.flush(buffer, oc);
+              await this.flushConversationMessagesBestEffort(persistQueue, buffer, oc);
             }
 
             const feedbackValue = (() => {
@@ -2217,7 +2217,7 @@ export class Agent {
                 });
 
                 if (!shouldDeferPersist && shouldPersistMemory) {
-                  await persistQueue.flush(buffer, oc);
+                  await this.flushConversationMessagesBestEffort(persistQueue, buffer, oc);
                 }
 
                 // History update removed - using OpenTelemetry only
@@ -4427,6 +4427,22 @@ export class Agent {
     }
   }
 
+  private async flushConversationMessagesBestEffort(
+    queue: MemoryPersistQueue,
+    buffer: ConversationBuffer,
+    oc: OperationContext,
+  ): Promise<void> {
+    try {
+      await queue.flush(buffer, oc);
+    } catch (error) {
+      oc.logger.debug("Failed to persist conversation messages", {
+        error,
+        conversationId: oc.conversationId,
+        userId: oc.userId,
+      });
+    }
+  }
+
   /**
    * Get contextual logger with parent tracking
    */

packages/core/src/agent/conversation-buffer.ts

@@ -7,9 +7,9 @@ import { convertModelMessagesToUIMessages } from "../utils/message-converter";
 
 type MessageSource = "user" | "system" | "memory" | "response";
 
-interface PendingMessage {
-  id: string;
+export interface PendingMessage {
   message: UIMessage;
+  version: number;
 }
 
 export interface ConversationBufferCheckpoint {
@@ -62,6 +62,8 @@ const extractOpenAIItemId = (metadata: unknown): string => {
 export class ConversationBuffer {
   private messages: UIMessage[] = [];
   private pendingMessageIds = new Set<string>();
+  private pendingMessageVersions = new Map<string, number>();
+  private nextPendingMessageVersion = 0;
   private toolPartIndex = new Map<string, { messageIndex: number; partIndex: number }>();
   private activeAssistantMessageId?: string;
 
@@ -128,6 +130,9 @@ export class ConversationBuffer {
   restoreCheckpoint(checkpoint: ConversationBufferCheckpoint): void {
     this.messages = checkpoint.messages.map((message) => this.cloneMessage(message));
     this.pendingMessageIds = new Set(checkpoint.pendingMessageIds);
+    this.pendingMessageVersions = new Map<string, number>(
+      checkpoint.pendingMessageIds.map((id) => [id, ++this.nextPendingMessageVersion]),
+    );
     this.activeAssistantMessageId = checkpoint.activeAssistantMessageId;
     this.rebuildToolPartIndex();
     this.log("restore-checkpoint", {
@@ -137,32 +142,53 @@ export class ConversationBuffer {
   }
 
   drainPendingMessages(): UIMessage[] {
+    const drained = this.getPendingMessages();
+    this.pendingMessageIds.clear();
+    this.pendingMessageVersions.clear();
+    if (
+      this.activeAssistantMessageId &&
+      drained.some((item) => item.message.id === this.activeAssistantMessageId)
+    ) {
+      this.activeAssistantMessageId = undefined;
+    }
+    if (drained.length > 0) {
+      this.log("drain-pending", {
+        count: drained.length,
+        ids: drained.map((item) => item.message.id),
+      });
+    }
+    return drained.map((item) => item.message);
+  }
+
+  getPendingMessages(): PendingMessage[] {
     if (this.pendingMessageIds.size === 0) {
       return [];
     }
 
-    const drained: PendingMessage[] = [];
+    const pending: PendingMessage[] = [];
 
     this.messages.forEach((message) => {
       if (this.pendingMessageIds.has(message.id)) {
-        drained.push({ id: message.id, message: this.cloneMessage(message) });
+        pending.push({
+          message: this.cloneMessage(message),
+          version: this.pendingMessageVersions.get(message.id) ?? 0,
+        });
       }
     });
 
-    this.pendingMessageIds.clear();
+    return pending;
+  }
 
-    if (drained.length > 0) {
-      const drainedIds = new Set(drained.map((item) => item.id));
-      if (this.activeAssistantMessageId && drainedIds.has(this.activeAssistantMessageId)) {
-        this.activeAssistantMessageId = undefined;
-      }
+  markMessagePersisted(id: string, version: number): void {
+    if (this.pendingMessageVersions.get(id) !== version) {
+      return;
     }
 
-    if (drained.length > 0) {
-      this.log("drain-pending", { count: drained.length, ids: drained.map((item) => item.id) });
+    this.pendingMessageIds.delete(id);
+    this.pendingMessageVersions.delete(id);
+    if (this.activeAssistantMessageId === id) {
+      this.activeAssistantMessageId = undefined;
     }
-
-    return drained.map((item) => item.message);
   }
 
   getAllMessages(): UIMessage[] {
@@ -197,7 +223,7 @@ export class ConversationBuffer {
       ...(existing as Record<string, unknown>),
       ...metadata,
     } as UIMessage["metadata"];
-    this.pendingMessageIds.add(target.id);
+    this.markPending(target.id);
     return true;
   }
 
@@ -211,7 +237,7 @@ export class ConversationBuffer {
     this.registerToolParts(this.messages.length - 1);
 
     if (!options.markAsSaved) {
-      this.pendingMessageIds.add(hydrated.id);
+      this.markPending(hydrated.id);
     }
 
     this.log("append-existing", {
@@ -314,7 +340,7 @@ export class ConversationBuffer {
     }
 
     if (modified) {
-      this.pendingMessageIds.add(target.id);
+      this.markPending(target.id);
       this.registerToolParts(lastAssistantIndex);
     }
   }
@@ -370,19 +396,25 @@ export class ConversationBuffer {
     const cloned = this.cloneMessage(message);
     this.ensureMessageId(cloned);
     this.messages.push(cloned);
-    this.pendingMessageIds.add(cloned.id);
+    this.markPending(cloned.id);
     this.registerToolParts(this.messages.length - 1);
     this.log("append-message", { messageId: cloned.id, role: cloned.role, source });
 
     if (source === "memory") {
       this.pendingMessageIds.delete(cloned.id);
+      this.pendingMessageVersions.delete(cloned.id);
     }
 
     if (source === "response") {
       this.activeAssistantMessageId = cloned.id;
     }
   }
 
+  private markPending(id: string): void {
+    this.pendingMessageIds.add(id);
+    this.pendingMessageVersions.set(id, ++this.nextPendingMessageVersion);
+  }
+
   private registerToolParts(messageIndex: number): void {
     const message = this.messages[messageIndex];
     for (let index = 0; index < message.parts.length; index++) {