Enhance continuation storage with conversation ID

src/AgentCore.Application/Conversation/Conversations.cs

@@ -235,9 +235,9 @@ public ValueTask DetachPrincipalAsync(string conversationId, string principalKey
         }
 
         /// <inheritdoc />
-        public ValueTask SaveContinuationAsync(string continuationId, JsonElement envelope, CancellationToken cancellationToken = default)
+        public ValueTask SaveContinuationAsync(string continuationId, string conversationId, JsonElement envelope, CancellationToken cancellationToken = default)
         {
-            return Store.SaveContinuationAsync(continuationId, envelope, cancellationToken);
+            return Store.SaveContinuationAsync(continuationId, conversationId, envelope, cancellationToken);
         }
 
         /// <inheritdoc />

src/AgentCore.Application/Conversation/Memory/InMemoryConversationContinuations.cs

@@ -8,22 +8,26 @@ namespace AgentCore.Application.Conversation.Memory
     /// </summary>
     internal sealed class InMemoryConversationContinuations
     {
-        private readonly Dictionary<string, JsonElement> _envelopes = [];
+        private readonly Dictionary<string, (string ConversationId, JsonElement Envelope, DateTimeOffset UpdatedAt)> _envelopes = [];
 
         /// <summary>Files one session envelope under one continuation id, replacing any envelope already there.</summary>
         /// <param name="continuationId">The continuation id: a conversation id or a response id.</param>
+        /// <param name="conversationId">The conversation the continuation belongs to.</param>
         /// <param name="envelope">The serialized session, as the agent wrote it.</param>
-        public void Save(string continuationId, JsonElement envelope)
+        /// <param name="updatedAt">When this write happened. A rewrite of the same id resets its retention clock.</param>
+        public void Save(string continuationId, string conversationId, JsonElement envelope, DateTimeOffset updatedAt)
         {
-            _envelopes[continuationId] = envelope.Clone();
+            _envelopes[continuationId] = (conversationId, envelope.Clone(), updatedAt);
         }
 
         /// <summary>Reads the envelope one continuation id names.</summary>
         /// <param name="continuationId">The continuation id to look up.</param>
         /// <returns>The envelope, or <see langword="null"/> when nothing is filed under that id.</returns>
         public JsonElement? Get(string continuationId)
         {
-            return _envelopes.TryGetValue(continuationId, out JsonElement envelope) ? envelope : null;
+            return _envelopes.TryGetValue(continuationId, out (string ConversationId, JsonElement Envelope, DateTimeOffset UpdatedAt) row)
+                ? row.Envelope
+                : null;
         }
 
         /// <summary>Withdraws whatever one continuation id names, if anything.</summary>
@@ -32,5 +36,38 @@ public void Forget(string continuationId)
         {
             _ = _envelopes.Remove(continuationId);
         }
+
+        /// <summary>Withdraws every continuation filed under one conversation, however it was keyed.</summary>
+        /// <param name="conversationId">The conversation whose continuations are gone.</param>
+        public void ForgetConversation(string conversationId)
+        {
+            foreach (string continuationId in _envelopes
+                .Where(pair => pair.Value.ConversationId == conversationId)
+                .Select(pair => pair.Key)
+                .ToList())
+            {
+                _ = _envelopes.Remove(continuationId);
+            }
+        }
+
+        /// <summary>
+        /// Withdraws every continuation row untouched since before the cutoff, whichever conversation
+        /// it belongs to.
+        /// </summary>
+        /// <param name="cutoff">The point in time a row must have been written on or after, to stay.</param>
+        /// <returns>How many rows went.</returns>
+        public int Sweep(DateTimeOffset cutoff)
+        {
+            List<string> going = [.. _envelopes
+                .Where(pair => pair.Value.UpdatedAt < cutoff)
+                .Select(pair => pair.Key)];
+
+            foreach (string continuationId in going)
+            {
+                _ = _envelopes.Remove(continuationId);
+            }
+
+            return going.Count;
+        }
     }
 }

src/AgentCore.Application/Conversation/Memory/InMemoryConversationStore.cs

@@ -147,22 +147,7 @@ public ValueTask<int> SweepAsync(
             {
                 DateTimeOffset cutoff = _time.GetUtcNow() - retention;
 
-                // SortValue is the clock ListAsync ranks by, so a conversation is swept exactly when it has
-                // fallen off the end of the list.
-                //
-                // batchSize is read for its guard and then ignored. It exists to keep one transaction
-                // short in a durable backing, and this store has no transaction; honouring it here would
-                // only make the caller loop for the same answer.
-                List<string> going = [.. _conversations.Values
-                    .Where(conversation => InMemoryConversationListing.SortValue(conversation) < cutoff)
-                    .Select(conversation => conversation.ConversationId)];
-
-                foreach (string? conversationId in going)
-                {
-                    Forget(conversationId);
-                }
-
-                return ValueTask.FromResult(going.Count);
+                return ValueTask.FromResult(_continuations.Sweep(cutoff));
             }
         }
 
@@ -311,13 +296,14 @@ public ValueTask<int> EraseAsync(string conversationId, CancellationToken cancel
         }
 
         /// <inheritdoc />
-        public ValueTask SaveContinuationAsync(string continuationId, JsonElement envelope, CancellationToken cancellationToken = default)
+        public ValueTask SaveContinuationAsync(string continuationId, string conversationId, JsonElement envelope, CancellationToken cancellationToken = default)
         {
             ArgumentNullException.ThrowIfNull(continuationId);
+            ArgumentNullException.ThrowIfNull(conversationId);
 
             lock (_lock)
             {
-                _continuations.Save(continuationId, envelope);
+                _continuations.Save(continuationId, conversationId, envelope, _time.GetUtcNow());
             }
 
             return default;
@@ -353,7 +339,7 @@ private void Forget(string conversationId)
             _ = _conversations.Remove(conversationId);
             _claims.Forget(conversationId);
             _ = _state.Remove(conversationId);
-            _continuations.Forget(conversationId);
+            _continuations.ForgetConversation(conversationId);
             _words.Forget(conversationId);
         }
 

src/AgentCore.Application/Ports/IConversationStore.cs

@@ -180,18 +180,19 @@ ValueTask<ConversationCut> TruncateAsync(
         /// <returns>How many messages went.</returns>
         ValueTask<int> EraseAsync(string conversationId, CancellationToken cancellationToken = default);
 
-        /// <summary>Deletes every conversation whose last activity is older than the retention window.</summary>
+        /// <summary>
+        /// Deletes every continuation row untouched past the retention window.
+        /// </summary>
         /// <param name="retention">
-        /// How long a conversation is kept, measured from its most recent message, or from when it was made when
-        /// it holds none. The window belongs to a deployment: it is not a schema key, and nothing here
-        /// defaults it.
+        /// How long a continuation row is kept, measured from when it was last written. The window belongs
+        /// to a deployment: it is not a schema key, and nothing here defaults it.
         /// </param>
         /// <param name="batchSize">
-        /// How many conversations one transaction may delete. The sweep loops until a batch deletes nothing, so
-        /// this bounds one transaction and never the work.
+        /// How many continuation rows one transaction may delete. The sweep loops until a batch deletes
+        /// nothing, so this bounds one transaction and never the work.
         /// </param>
         /// <param name="cancellationToken">Cancels the sweep between batches, and inside one.</param>
-        /// <returns>How many conversations went, over every batch.</returns>
+        /// <returns>How many continuation rows went, over every batch.</returns>
         ValueTask<int> SweepAsync(
             TimeSpan retention,
             int batchSize = 500,
@@ -219,6 +220,7 @@ ValueTask DetachPrincipalAsync(
 
         /// <summary>Files one session envelope under one continuation id, replacing any envelope already there.</summary>
         /// <param name="continuationId">The continuation id: a conversation id or a response id.</param>
+        /// <param name="conversationId">The conversation the continuation belongs to.</param>
         /// <param name="envelope">The serialized session, as the agent wrote it.</param>
         /// <param name="cancellationToken">Cancels the write.</param>
         /// <remarks>
@@ -227,6 +229,7 @@ ValueTask DetachPrincipalAsync(
         /// </remarks>
         ValueTask SaveContinuationAsync(
             string continuationId,
+            string conversationId,
             JsonElement envelope,
             CancellationToken cancellationToken = default);
 

src/AgentCore.Application/Ports/IConversations.cs

@@ -6,11 +6,7 @@
 namespace AgentCore.Application.Ports
 {
     /// <summary>
-    /// The door a host or a consumer reads and writes stored conversations through: rows, words, and the
-    /// files a conversation published, together. Words are read a window of turns at a time and never
-    /// whole: a consumer that could read a whole conversation would, and a long one is a slow query and a
-    /// large body every time. What the turn loop and the sweeper need beyond this stays on
-    /// <see cref="IConversationStore"/>, which is the adapter seam and not a consumer surface.
+    /// The door a host or a consumer reads and writes stored conversations through.
     /// </summary>
     public interface IConversations
     {
@@ -66,5 +62,11 @@ ValueTask<ConversationPage> ListAsync(
 
         /// <inheritdoc cref="IConversationStore.DetachPrincipalAsync"/>
         ValueTask DetachPrincipalAsync(string conversationId, string principalKey, CancellationToken cancellationToken = default);
+
+        /// <inheritdoc cref="IConversationStore.SweepAsync"/>
+        ValueTask<int> SweepAsync(
+            TimeSpan retention,
+            int batchSize = 500,
+            CancellationToken cancellationToken = default);
     }
 }

src/AgentCore.Application/PublicAPI.Unshipped.txt

@@ -140,7 +140,7 @@ AgentCore.Application.Conversation.Memory.InMemoryConversationStore.InMemoryConv
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.ListAsync(string! principalKey, string? after, int limit, AgentCore.Application.Conversation.ConversationStatus? status = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<AgentCore.Application.Conversation.ConversationPage!>
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.RenameAsync(string! conversationId, string! title, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.RewriteAsync(string! conversationId, string! messageId, Microsoft.Extensions.AI.ChatMessage! content, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
-AgentCore.Application.Conversation.Memory.InMemoryConversationStore.SaveContinuationAsync(string! continuationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
+AgentCore.Application.Conversation.Memory.InMemoryConversationStore.SaveContinuationAsync(string! continuationId, string! conversationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.SetCustomAsync(string! conversationId, System.Text.Json.JsonElement? custom, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.SetExternalIdAsync(string! conversationId, string? externalId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Memory.InMemoryConversationStore.SetStatusAsync(string! conversationId, AgentCore.Application.Conversation.ConversationStatus status, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
@@ -1076,7 +1076,8 @@ AgentCore.Application.Ports.IConversationStore.GetContinuationAsync(string! cont
 AgentCore.Application.Ports.IConversationStore.ListAsync(string! principalKey, string? after, int limit, AgentCore.Application.Conversation.ConversationStatus? status = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<AgentCore.Application.Conversation.ConversationPage!>
 AgentCore.Application.Ports.IConversationStore.RenameAsync(string! conversationId, string! title, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Ports.IConversationStore.RewriteAsync(string! conversationId, string! messageId, Microsoft.Extensions.AI.ChatMessage! content, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
-AgentCore.Application.Ports.IConversationStore.SaveContinuationAsync(string! continuationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
+AgentCore.Application.Ports.IConversationStore.SaveContinuationAsync(string! continuationId, string! conversationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
+AgentCore.Application.Ports.IConversations.SweepAsync(System.TimeSpan retention, int batchSize = 500, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<int>
 AgentCore.Application.Ports.IConversationStore.SetCustomAsync(string! conversationId, System.Text.Json.JsonElement? custom, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Ports.IConversationStore.SetExternalIdAsync(string! conversationId, string? externalId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Ports.IConversationStore.SetStatusAsync(string! conversationId, AgentCore.Application.Conversation.ConversationStatus status, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
@@ -2158,7 +2159,7 @@ AgentCore.Application.Conversation.Conversations.LinkFilesAsync(string! conversa
 AgentCore.Application.Conversation.Conversations.ListAsync(string! principalKey, string? after, int limit, AgentCore.Application.Conversation.ConversationStatus? status = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<AgentCore.Application.Conversation.ConversationPage!>
 AgentCore.Application.Conversation.Conversations.RenameAsync(string! conversationId, string! title, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Conversations.RewriteAsync(string! conversationId, string! messageId, Microsoft.Extensions.AI.ChatMessage! content, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
-AgentCore.Application.Conversation.Conversations.SaveContinuationAsync(string! continuationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
+AgentCore.Application.Conversation.Conversations.SaveContinuationAsync(string! continuationId, string! conversationId, System.Text.Json.JsonElement envelope, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Conversations.SetCustomAsync(string! conversationId, System.Text.Json.JsonElement? custom, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Conversations.SetExternalIdAsync(string! conversationId, string? externalId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
 AgentCore.Application.Conversation.Conversations.SetStatusAsync(string! conversationId, AgentCore.Application.Conversation.ConversationStatus status, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask

src/AgentCore.AspNetCore/Conversation/ConversationSweepOptions.cs

@@ -0,0 +1,15 @@
+namespace AgentCore.AspNetCore.Conversation
+{
+    /// <summary>Settings for <see cref="ConversationSweeper"/>.</summary>
+    public sealed class ConversationSweepOptions
+    {
+        /// <summary>How long a conversation is kept after its last activity, before the sweep erases it.</summary>
+        public TimeSpan Retention { get; set; } = TimeSpan.FromDays(90);
+
+        /// <summary>How often the sweep runs.</summary>
+        public TimeSpan Interval { get; set; } = TimeSpan.FromHours(1);
+
+        /// <summary>How many conversations one sweep pass erases at most.</summary>
+        public int BatchSize { get; set; } = 500;
+    }
+}