diff --git a/src/AgentCore.Application/Conversation/Conversations.cs b/src/AgentCore.Application/Conversation/Conversations.cs index 0f75cebc..a65e6153 100644 --- a/src/AgentCore.Application/Conversation/Conversations.cs +++ b/src/AgentCore.Application/Conversation/Conversations.cs @@ -235,9 +235,9 @@ public ValueTask DetachPrincipalAsync(string conversationId, string principalKey } /// - 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); } /// diff --git a/src/AgentCore.Application/Conversation/Memory/InMemoryConversationContinuations.cs b/src/AgentCore.Application/Conversation/Memory/InMemoryConversationContinuations.cs index 0d882986..b3897e1e 100644 --- a/src/AgentCore.Application/Conversation/Memory/InMemoryConversationContinuations.cs +++ b/src/AgentCore.Application/Conversation/Memory/InMemoryConversationContinuations.cs @@ -8,14 +8,16 @@ namespace AgentCore.Application.Conversation.Memory /// internal sealed class InMemoryConversationContinuations { - private readonly Dictionary _envelopes = []; + private readonly Dictionary _envelopes = []; /// Files one session envelope under one continuation id, replacing any envelope already there. /// The continuation id: a conversation id or a response id. + /// The conversation the continuation belongs to. /// The serialized session, as the agent wrote it. - public void Save(string continuationId, JsonElement envelope) + /// When this write happened. A rewrite of the same id resets its retention clock. + public void Save(string continuationId, string conversationId, JsonElement envelope, DateTimeOffset updatedAt) { - _envelopes[continuationId] = envelope.Clone(); + _envelopes[continuationId] = (conversationId, envelope.Clone(), updatedAt); } /// Reads the envelope one continuation id names. @@ -23,7 +25,9 @@ public void Save(string continuationId, JsonElement envelope) /// The envelope, or when nothing is filed under that id. 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; } /// Withdraws whatever one continuation id names, if anything. @@ -32,5 +36,38 @@ public void Forget(string continuationId) { _ = _envelopes.Remove(continuationId); } + + /// Withdraws every continuation filed under one conversation, however it was keyed. + /// The conversation whose continuations are gone. + public void ForgetConversation(string conversationId) + { + foreach (string continuationId in _envelopes + .Where(pair => pair.Value.ConversationId == conversationId) + .Select(pair => pair.Key) + .ToList()) + { + _ = _envelopes.Remove(continuationId); + } + } + + /// + /// Withdraws every continuation row untouched since before the cutoff, whichever conversation + /// it belongs to. + /// + /// The point in time a row must have been written on or after, to stay. + /// How many rows went. + public int Sweep(DateTimeOffset cutoff) + { + List going = [.. _envelopes + .Where(pair => pair.Value.UpdatedAt < cutoff) + .Select(pair => pair.Key)]; + + foreach (string continuationId in going) + { + _ = _envelopes.Remove(continuationId); + } + + return going.Count; + } } } diff --git a/src/AgentCore.Application/Conversation/Memory/InMemoryConversationStore.cs b/src/AgentCore.Application/Conversation/Memory/InMemoryConversationStore.cs index 51cc5dd0..553ec0d8 100644 --- a/src/AgentCore.Application/Conversation/Memory/InMemoryConversationStore.cs +++ b/src/AgentCore.Application/Conversation/Memory/InMemoryConversationStore.cs @@ -147,22 +147,7 @@ public ValueTask 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 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 EraseAsync(string conversationId, CancellationToken cancel } /// - 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); } diff --git a/src/AgentCore.Application/Ports/IConversationStore.cs b/src/AgentCore.Application/Ports/IConversationStore.cs index 6fa76ca0..3ea1049a 100644 --- a/src/AgentCore.Application/Ports/IConversationStore.cs +++ b/src/AgentCore.Application/Ports/IConversationStore.cs @@ -180,18 +180,19 @@ ValueTask TruncateAsync( /// How many messages went. ValueTask EraseAsync(string conversationId, CancellationToken cancellationToken = default); - /// Deletes every conversation whose last activity is older than the retention window. + /// + /// Deletes every continuation row untouched past the retention window. + /// /// - /// 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. /// /// - /// 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. /// /// Cancels the sweep between batches, and inside one. - /// How many conversations went, over every batch. + /// How many continuation rows went, over every batch. ValueTask SweepAsync( TimeSpan retention, int batchSize = 500, @@ -219,6 +220,7 @@ ValueTask DetachPrincipalAsync( /// Files one session envelope under one continuation id, replacing any envelope already there. /// The continuation id: a conversation id or a response id. + /// The conversation the continuation belongs to. /// The serialized session, as the agent wrote it. /// Cancels the write. /// @@ -227,6 +229,7 @@ ValueTask DetachPrincipalAsync( /// ValueTask SaveContinuationAsync( string continuationId, + string conversationId, JsonElement envelope, CancellationToken cancellationToken = default); diff --git a/src/AgentCore.Application/Ports/IConversations.cs b/src/AgentCore.Application/Ports/IConversations.cs index d8648a48..89ec85af 100644 --- a/src/AgentCore.Application/Ports/IConversations.cs +++ b/src/AgentCore.Application/Ports/IConversations.cs @@ -6,11 +6,7 @@ namespace AgentCore.Application.Ports { /// - /// 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 - /// , which is the adapter seam and not a consumer surface. + /// The door a host or a consumer reads and writes stored conversations through. /// public interface IConversations { @@ -66,5 +62,11 @@ ValueTask ListAsync( /// ValueTask DetachPrincipalAsync(string conversationId, string principalKey, CancellationToken cancellationToken = default); + + /// + ValueTask SweepAsync( + TimeSpan retention, + int batchSize = 500, + CancellationToken cancellationToken = default); } } diff --git a/src/AgentCore.Application/PublicAPI.Unshipped.txt b/src/AgentCore.Application/PublicAPI.Unshipped.txt index 6a65cfd8..a250e84d 100644 --- a/src/AgentCore.Application/PublicAPI.Unshipped.txt +++ b/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.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.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 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.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 diff --git a/src/AgentCore.AspNetCore/Conversation/ConversationSweepOptions.cs b/src/AgentCore.AspNetCore/Conversation/ConversationSweepOptions.cs new file mode 100644 index 00000000..6ad3138f --- /dev/null +++ b/src/AgentCore.AspNetCore/Conversation/ConversationSweepOptions.cs @@ -0,0 +1,15 @@ +namespace AgentCore.AspNetCore.Conversation +{ + /// Settings for . + public sealed class ConversationSweepOptions + { + /// How long a conversation is kept after its last activity, before the sweep erases it. + public TimeSpan Retention { get; set; } = TimeSpan.FromDays(90); + + /// How often the sweep runs. + public TimeSpan Interval { get; set; } = TimeSpan.FromHours(1); + + /// How many conversations one sweep pass erases at most. + public int BatchSize { get; set; } = 500; + } +} diff --git a/src/AgentCore.AspNetCore/Conversation/ConversationSweeper.cs b/src/AgentCore.AspNetCore/Conversation/ConversationSweeper.cs new file mode 100644 index 00000000..f71babca --- /dev/null +++ b/src/AgentCore.AspNetCore/Conversation/ConversationSweeper.cs @@ -0,0 +1,67 @@ +using AgentCore.Application.Ports; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace AgentCore.AspNetCore.Conversation +{ + /// + /// Runs the retention sweep of for as long as the host is up. + /// + /// + /// AgentCore never registers this on its own: a host that wants old conversations erased opts in by + /// calling AddConversationSweep. Without that call, conversations are kept forever. + /// + public sealed class ConversationSweeper( + IConversations conversations, + IOptions options, + TimeProvider timeProvider, + ILogger logger) : BackgroundService + { + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + ConversationSweepOptions settings = options.Value; + + using PeriodicTimer timer = new(settings.Interval, timeProvider); + + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) + { + try + { + int swept = await conversations + .SweepAsync(settings.Retention, settings.BatchSize, stoppingToken) + .ConfigureAwait(false); + + ConversationSweeperLog.Swept(logger, swept); + } + catch (Exception fault) when (fault is not OperationCanceledException) + { + ConversationSweeperLog.SweepFaulted(logger, fault); + } + } + } + } + + /// Every line the retention sweep writes. + internal static partial class ConversationSweeperLog + { + /// One sweep pass finished. + /// The logger of the sweeper. + /// How many conversations the pass erased. + [LoggerMessage( + EventId = 1, + Level = LogLevel.Debug, + Message = "the retention sweep erased {Count} conversation(s).")] + public static partial void Swept(ILogger logger, int count); + + /// A sweep pass could not finish. + /// The logger of the sweeper. + /// The cause. + [LoggerMessage( + EventId = 2, + Level = LogLevel.Error, + Message = "the retention sweep faulted, so this pass erased nothing. It will retry on the next tick.")] + public static partial void SweepFaulted(ILogger logger, Exception exception); + } +} diff --git a/src/AgentCore.AspNetCore/DependencyInjection/AgentCoreServiceCollectionExtensions.cs b/src/AgentCore.AspNetCore/DependencyInjection/AgentCoreServiceCollectionExtensions.cs index 9c31dc91..956c4802 100644 --- a/src/AgentCore.AspNetCore/DependencyInjection/AgentCoreServiceCollectionExtensions.cs +++ b/src/AgentCore.AspNetCore/DependencyInjection/AgentCoreServiceCollectionExtensions.cs @@ -4,6 +4,7 @@ using AgentCore.Application.Configuration.Schema; using AgentCore.Application.Evaluation; using AgentCore.Application.Ports; +using AgentCore.AspNetCore.Conversation; using AgentCore.AspNetCore.DependencyInjection.Startup; using AgentCore.AspNetCore.Sessions; using Microsoft.Agents.AI.Hosting; @@ -98,6 +99,30 @@ public static IServiceCollection AddAgentCore( return services; } + /// + /// Opts this host into the periodic retention sweep of stored conversations. + /// + /// The service collection of the host. + /// Sets the retention, the interval, and the batch size. Defaults live on . + /// The same collection, so a host chains its conversations. + public static IServiceCollection AddConversationSweep( + this IServiceCollection services, + Action? configure = null) + { + ArgumentNullException.ThrowIfNull(services); + + OptionsBuilder builder = services.AddOptions(); + + if (configure is not null) + { + _ = builder.Configure(configure); + } + + _ = services.AddHostedService(); + + return services; + } + /// Registers WebSocket options that suit a phone conversation rather than a browser tab. /// The service collection of the host. /// The same collection, so a host chains its conversations. diff --git a/src/AgentCore.AspNetCore/Endpoints/ResponsesSessionFiling.cs b/src/AgentCore.AspNetCore/Endpoints/ResponsesSessionFiling.cs index 197aa9d6..522329e6 100644 --- a/src/AgentCore.AspNetCore/Endpoints/ResponsesSessionFiling.cs +++ b/src/AgentCore.AspNetCore/Endpoints/ResponsesSessionFiling.cs @@ -6,21 +6,33 @@ namespace AgentCore.AspNetCore.Endpoints /// Files a session the Responses path just advanced, under every id that names it. internal static class ResponsesSessionFiling { - /// Files the session the turn just advanced, under every id that names it now. + /// + /// Files the session the turn just advanced, under every id that names it now. The response id + /// always gets the frozen, full envelope; the conversation id gets a pointer at it, so the two + /// never hold two copies of the same state. + /// + /// The store to file the envelope in. + /// The agent the session belongs to. + /// The session the turn advanced. + /// The id minted for this turn's answer. + /// The client-facing conversation id, or when none names it. + /// The internal conversation the continuation rows belong to, always present. + /// Cancels the write. internal static async Task SaveAsync( AgentCoreAgentSessionStore sessions, AIAgent agent, AgentSession session, string responseId, string? conversationId, + string ownerConversationId, CancellationToken cancellationToken) { + await sessions.SaveSessionForConversationAsync(agent, responseId, ownerConversationId, session, cancellationToken).ConfigureAwait(false); + if (conversationId is not null) { - await sessions.SaveSessionAsync(agent, conversationId, session, cancellationToken).ConfigureAwait(false); + await sessions.SavePointerAsync(conversationId, ownerConversationId, responseId, cancellationToken).ConfigureAwait(false); } - - await sessions.SaveSessionAsync(agent, responseId, session, cancellationToken).ConfigureAwait(false); } } } diff --git a/src/AgentCore.AspNetCore/Endpoints/ResponsesTurn.cs b/src/AgentCore.AspNetCore/Endpoints/ResponsesTurn.cs index fae61c91..4d8e8735 100644 --- a/src/AgentCore.AspNetCore/Endpoints/ResponsesTurn.cs +++ b/src/AgentCore.AspNetCore/Endpoints/ResponsesTurn.cs @@ -27,7 +27,8 @@ internal sealed record ResponsesTurn( /// Files the session, under every id that names it now. public Task FileAsync(CancellationToken cancellationToken) { - return ResponsesSessionFiling.SaveAsync(Sessions, Agent, Session, ResponseId, ConversationId, cancellationToken); + return ResponsesSessionFiling.SaveAsync( + Sessions, Agent, Session, ResponseId, ConversationId, Conversation.ConversationId, cancellationToken); } } } diff --git a/src/AgentCore.AspNetCore/PublicAPI.Unshipped.txt b/src/AgentCore.AspNetCore/PublicAPI.Unshipped.txt index 23db8b54..7b434542 100644 --- a/src/AgentCore.AspNetCore/PublicAPI.Unshipped.txt +++ b/src/AgentCore.AspNetCore/PublicAPI.Unshipped.txt @@ -56,6 +56,8 @@ AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.ContainsAsync(string! s override AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.DeleteSessionAsync(Microsoft.Agents.AI.AIAgent! agent, string! sessionStoreId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask override AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.GetSessionAsync(Microsoft.Agents.AI.AIAgent! agent, string! sessionStoreId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask override AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.SaveSessionAsync(Microsoft.Agents.AI.AIAgent! agent, string! sessionStoreId, Microsoft.Agents.AI.AgentSession! session, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.SaveSessionForConversationAsync(Microsoft.Agents.AI.AIAgent! agent, string! sessionStoreId, string! conversationId, Microsoft.Agents.AI.AgentSession! session, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask +AgentCore.AspNetCore.Sessions.AgentCoreAgentSessionStore.SavePointerAsync(string! sessionStoreId, string! conversationId, string! target, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask AgentCore.AspNetCore.Vendors.TelnyxRelay.TelnyxRelayConversationAdapter AgentCore.AspNetCore.Vendors.TelnyxRelay.TelnyxRelayConversationAdapter.CarriesText.get -> bool AgentCore.AspNetCore.Vendors.TelnyxRelay.TelnyxRelayConversationAdapter.CreateHandler(AgentCore.Application.Configuration.Schema.ConversationProviderConfiguration! configuration) -> Microsoft.AspNetCore.Http.RequestDelegate! @@ -85,3 +87,14 @@ static readonly AgentCore.AspNetCore.DependencyInjection.AgentCoreHttpClients.Fi static readonly AgentCore.AspNetCore.DependencyInjection.AgentCoreHttpClients.RequestDeadline -> System.TimeSpan AgentCore.AspNetCore.DependencyInjection.AgentCoreOptions.UseBlobStores(params AgentCore.Application.Ports.IBlobStoreAdapter![]! adapters) -> AgentCore.AspNetCore.DependencyInjection.AgentCoreOptions! const AgentCore.AspNetCore.Endpoints.ResponsesEndpointRouteBuilderExtensions.TimeZoneHeaderName = "X-AgentCore-Time-Zone" -> string! +AgentCore.AspNetCore.Conversation.ConversationSweepOptions +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.ConversationSweepOptions() -> void +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.Retention.get -> System.TimeSpan +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.Retention.set -> void +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.Interval.get -> System.TimeSpan +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.Interval.set -> void +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.BatchSize.get -> int +AgentCore.AspNetCore.Conversation.ConversationSweepOptions.BatchSize.set -> void +AgentCore.AspNetCore.Conversation.ConversationSweeper +AgentCore.AspNetCore.Conversation.ConversationSweeper.ConversationSweeper(AgentCore.Application.Ports.IConversations! conversations, Microsoft.Extensions.Options.IOptions! options, System.TimeProvider! timeProvider, Microsoft.Extensions.Logging.ILogger! logger) -> void +static AgentCore.AspNetCore.DependencyInjection.AgentCoreServiceCollectionExtensions.AddConversationSweep(this Microsoft.Extensions.DependencyInjection.IServiceCollection! services, System.Action? configure = null) -> Microsoft.Extensions.DependencyInjection.IServiceCollection! diff --git a/src/AgentCore.AspNetCore/Sessions/AgentCoreAgentSessionStore.cs b/src/AgentCore.AspNetCore/Sessions/AgentCoreAgentSessionStore.cs index d2ee08f9..93ed2933 100644 --- a/src/AgentCore.AspNetCore/Sessions/AgentCoreAgentSessionStore.cs +++ b/src/AgentCore.AspNetCore/Sessions/AgentCoreAgentSessionStore.cs @@ -10,6 +10,8 @@ namespace AgentCore.AspNetCore.Sessions /// public sealed class AgentCoreAgentSessionStore : AgentSessionStore { + private const string PointerProperty = "$continuationPointer"; + private readonly IConversationStore _conversations; /// Creates the seam over one conversation store. @@ -34,17 +36,52 @@ public async ValueTask ContainsAsync(string sessionStoreId, CancellationTo } /// - public override async ValueTask SaveSessionAsync( + public override ValueTask SaveSessionAsync( AIAgent agent, string sessionStoreId, AgentSession session, CancellationToken cancellationToken = default) + { + return SaveSessionForConversationAsync(agent, sessionStoreId, sessionStoreId, session, cancellationToken); + } + + /// Files the session under one continuation id, on behalf of one conversation. + /// The agent the session belongs to. + /// The conversation id or response id to file the envelope under. + /// The conversation the continuation belongs to. + /// The session to file. + /// Cancels the write. + public async ValueTask SaveSessionForConversationAsync( + AIAgent agent, string sessionStoreId, string conversationId, AgentSession session, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(agent); ArgumentNullException.ThrowIfNull(sessionStoreId); + ArgumentNullException.ThrowIfNull(conversationId); ArgumentNullException.ThrowIfNull(session); JsonElement envelope = await agent .SerializeSessionAsync(session, cancellationToken: cancellationToken) .ConfigureAwait(false); - await _conversations.SaveContinuationAsync(sessionStoreId, envelope, cancellationToken).ConfigureAwait(false); + + await _conversations.SaveContinuationAsync(sessionStoreId, conversationId, envelope, cancellationToken).ConfigureAwait(false); + } + + /// + /// Points one continuation id at another, in place of a full envelope. A resume through the + /// pointer id always reads the latest state, without a second copy of it. + /// + /// The continuation id the pointer is filed under. + /// The conversation the pointer belongs to. + /// The continuation id holding the full envelope. + /// Cancels the write. + public ValueTask SavePointerAsync( + string sessionStoreId, string conversationId, string target, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(sessionStoreId); + ArgumentNullException.ThrowIfNull(conversationId); + ArgumentNullException.ThrowIfNull(target); + + JsonElement pointer = JsonSerializer.SerializeToElement( + new Dictionary { [PointerProperty] = target }); + + return _conversations.SaveContinuationAsync(sessionStoreId, conversationId, pointer, cancellationToken); } /// @@ -55,13 +92,39 @@ public override async ValueTask GetSessionAsync( ArgumentNullException.ThrowIfNull(sessionStoreId); JsonElement? envelope = await _conversations.GetContinuationAsync(sessionStoreId, cancellationToken).ConfigureAwait(false); - return envelope is null - ? await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false) - : await agent + + if (envelope is null) + { + return await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); + } + + if (TryReadPointerTarget(envelope.Value, out string? target)) + { + envelope = await _conversations.GetContinuationAsync(target, cancellationToken).ConfigureAwait(false) + ?? throw new InvalidOperationException( + $"Continuation '{sessionStoreId}' points at '{target}', which no longer names an envelope."); + } + + return await agent .DeserializeSessionAsync(envelope.Value, cancellationToken: cancellationToken) .ConfigureAwait(false); } + /// Reads the target a pointer row names, or answers for a full envelope. + private static bool TryReadPointerTarget(JsonElement envelope, out string target) + { + if (envelope.ValueKind == JsonValueKind.Object + && envelope.TryGetProperty(PointerProperty, out JsonElement value) + && value.ValueKind == JsonValueKind.String) + { + target = value.GetString()!; + return true; + } + + target = ""; + return false; + } + /// public override ValueTask DeleteSessionAsync( AIAgent agent, string sessionStoreId, CancellationToken cancellationToken = default) diff --git a/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStore.cs b/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStore.cs index a225a2ae..5430dc5f 100644 --- a/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStore.cs +++ b/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStore.cs @@ -109,10 +109,6 @@ public async ValueTask DeleteAsync(string conversationId, CancellationToken canc await using NpgsqlCommand command = _dataSource.CreateCommand(DeleteSql); _ = command.Parameters.Add(new NpgsqlParameter { Value = conversationId }); _ = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); - - await using NpgsqlCommand continuation = _dataSource.CreateCommand(DeleteContinuationSql); - _ = continuation.Parameters.Add(new NpgsqlParameter { Value = conversationId }); - _ = await continuation.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); } /// @@ -177,15 +173,7 @@ public async ValueTask SweepAsync( while (true) { - await using NpgsqlCommand continuations = _dataSource.CreateCommand(SweepContinuationsSql); - - _ = continuations.Parameters.Add( - new NpgsqlParameter { Value = retention, NpgsqlDbType = NpgsqlDbType.Interval }); - - _ = continuations.Parameters.Add(new NpgsqlParameter { Value = batchSize }); - _ = await continuations.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); - - await using NpgsqlCommand command = _dataSource.CreateCommand(SweepSql); + await using NpgsqlCommand command = _dataSource.CreateCommand(SweepContinuationsSql); _ = command.Parameters.Add( new NpgsqlParameter { Value = retention, NpgsqlDbType = NpgsqlDbType.Interval }); @@ -298,12 +286,14 @@ public ValueTask EraseAsync(string conversationId, CancellationToken cancel } /// - public async ValueTask SaveContinuationAsync(string continuationId, JsonElement envelope, CancellationToken cancellationToken = default) + public async ValueTask SaveContinuationAsync(string continuationId, string conversationId, JsonElement envelope, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(continuationId); + ArgumentNullException.ThrowIfNull(conversationId); await using NpgsqlCommand command = _dataSource.CreateCommand(SaveContinuationSql); _ = command.Parameters.Add(new NpgsqlParameter { Value = continuationId }); + _ = command.Parameters.Add(new NpgsqlParameter { Value = conversationId }); _ = command.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Jsonb, diff --git a/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStoreSql.cs b/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStoreSql.cs index 157d815b..675af5bd 100644 --- a/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStoreSql.cs +++ b/src/AgentCore.Infrastructure/Conversation/Postgres/PostgresConversationStoreSql.cs @@ -175,8 +175,8 @@ SELECT ordinal FROM {Schema}.conversation_message /// Files one session envelope under one continuation id, replacing any envelope already there. internal const string SaveContinuationSql = $""" - INSERT INTO {Schema}.response_continuation (store_id, envelope) VALUES ($1, $2) - ON CONFLICT (store_id) DO UPDATE SET envelope = EXCLUDED.envelope + INSERT INTO {Schema}.response_continuation (store_id, conversation_id, envelope) VALUES ($1, $2, $3) + ON CONFLICT (store_id) DO UPDATE SET envelope = EXCLUDED.envelope, updated_at = now() """; /// Reads the envelope one continuation id names. @@ -216,32 +216,19 @@ FROM spoken m JOIN completed a USING (conversation_id, turn_index) ORDER BY m.turn_index """; - /// Deletes one batch of conversations that have aged out whole. + /// + /// Deletes one batch of continuation rows that have gone untouched past the retention window. + /// Conversations are never swept: a chat id and every response id it ever advanced age out + /// on their own, independently of whether the conversation itself is still open. + /// internal static readonly string SweepContinuationsSql = $""" DELETE FROM {Schema}.response_continuation WHERE store_id IN ( - SELECT conversation_id FROM ( - SELECT c.conversation_id - FROM {Schema}.conversation c - {ActivityJoin} - WHERE {SortAt} < now() - $1 - LIMIT $2 - ) q) - """; - - /// Deletes one batch of conversations that have aged out whole. - internal static readonly string SweepSql = - $""" - DELETE FROM {Schema}.conversation - WHERE conversation_id IN ( - SELECT conversation_id FROM ( - SELECT c.conversation_id - FROM {Schema}.conversation c - {ActivityJoin} - WHERE {SortAt} < now() - $1 - LIMIT $2 - ) q) + SELECT store_id FROM {Schema}.response_continuation + WHERE updated_at < now() - $1 + LIMIT $2 + ) """; } } diff --git a/src/AgentCore.Infrastructure/Database/Postgres/Migrations/001_agentcore.sql b/src/AgentCore.Infrastructure/Database/Postgres/Migrations/001_agentcore.sql index 98d322b7..5c10d948 100644 --- a/src/AgentCore.Infrastructure/Database/Postgres/Migrations/001_agentcore.sql +++ b/src/AgentCore.Infrastructure/Database/Postgres/Migrations/001_agentcore.sql @@ -65,11 +65,17 @@ CREATE INDEX conversation_message_summary_idx ON agentcore.conversation_messa GRANT SELECT, INSERT, UPDATE, DELETE ON agentcore.conversation_message TO agentcore_writer; CREATE TABLE agentcore.response_continuation ( - store_id text PRIMARY KEY, - envelope jsonb NOT NULL, - created_at timestamptz NOT NULL DEFAULT now() + store_id text PRIMARY KEY, + conversation_id text NOT NULL REFERENCES agentcore.conversation(conversation_id) ON DELETE CASCADE, + envelope jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() ); +CREATE INDEX response_continuation_conversation_idx ON agentcore.response_continuation (conversation_id); + +CREATE INDEX response_continuation_retention_idx ON agentcore.response_continuation (updated_at); + REVOKE ALL ON agentcore.response_continuation FROM PUBLIC; REVOKE ALL ON agentcore.response_continuation FROM agentcore_writer; diff --git a/tests/AgentCore.Application.Tests/Conversation/InMemoryConversationStoreTests.cs b/tests/AgentCore.Application.Tests/Conversation/InMemoryConversationStoreTests.cs index 37d0f01b..c0c4263c 100644 --- a/tests/AgentCore.Application.Tests/Conversation/InMemoryConversationStoreTests.cs +++ b/tests/AgentCore.Application.Tests/Conversation/InMemoryConversationStoreTests.cs @@ -255,7 +255,7 @@ public async Task DeleteAsync_AConversation_TakesItsSameIdContinuationWithIt() InMemoryConversationStore store = new(); _ = await store.CreateAsync("c1", Token); using JsonDocument document = JsonDocument.Parse("""{ "conversationId": "c1" }"""); - await store.SaveContinuationAsync("c1", document.RootElement, Token); + await store.SaveContinuationAsync("c1", "c1", document.RootElement, Token); // Act await store.DeleteAsync("c1", Token); @@ -265,14 +265,17 @@ public async Task DeleteAsync_AConversation_TakesItsSameIdContinuationWithIt() } [Fact] - public async Task SweepAsync_AConversationPastRetention_TakesItsResumeStateWithIt() + public async Task SweepAsync_AContinuationPastRetention_TakesTheRowOnly() { - // Arrange — retention is the promise that a conversation stops existing, and slots hold what the - // extractor took from the caller, so state left behind is data kept past the promise. + // Arrange — retention ages out resume state, not the conversation it resumes: the words and + // the slots the extractor filled stay, the same way a stale response id stops resolving with + // most providers while the conversation itself lives on. TestTimeProvider clock = new(); InMemoryConversationStore store = new(clock); _ = await store.CreateAsync("c1", Token); _ = await store.AppendAsync("c1", [Word()], new ConversationSessionState { Stage = "collecting" }, Token); + using JsonDocument document = JsonDocument.Parse("""{ "state": {} }"""); + await store.SaveContinuationAsync("c1", "c1", document.RootElement, Token); // Act clock.Advance(TimeSpan.FromDays(2)); @@ -280,7 +283,8 @@ public async Task SweepAsync_AConversationPastRetention_TakesItsResumeStateWithI // Assert Assert.Equal(1, swept); - Assert.Null((await store.CreateAsync("c1", Token)).State); + Assert.Null(await store.GetContinuationAsync("c1", Token)); + Assert.NotNull(await store.GetAsync("c1", Token)); } private static ConversationMessageDraft Word() diff --git a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/ContinuationStoreTests.cs b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/ContinuationStoreTests.cs index a226dfad..8cca20ac 100644 --- a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/ContinuationStoreTests.cs +++ b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/ContinuationStoreTests.cs @@ -30,10 +30,11 @@ public async Task RoundTrip_SaveGetDelete() { // Arrange PostgresConversationStore store = new(DataSource); + _ = await store.CreateAsync("conv_1", Token); using JsonDocument document = JsonDocument.Parse("""{ "conversationId": "c1", "state": {} }"""); // Act - await store.SaveContinuationAsync("conv_1", document.RootElement, Token); + await store.SaveContinuationAsync("conv_1", "conv_1", document.RootElement, Token); JsonElement? found = await store.GetContinuationAsync("conv_1", Token); // Assert @@ -42,7 +43,7 @@ public async Task RoundTrip_SaveGetDelete() // Act using JsonDocument replacement = JsonDocument.Parse("{}"); - await store.SaveContinuationAsync("conv_1", replacement.RootElement, Token); + await store.SaveContinuationAsync("conv_1", "conv_1", replacement.RootElement, Token); JsonElement? replaced = await store.GetContinuationAsync("conv_1", Token); // Assert diff --git a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreSweepTests.cs b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreSweepTests.cs index 279ecaa7..de8b8df1 100644 --- a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreSweepTests.cs +++ b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreSweepTests.cs @@ -1,3 +1,4 @@ +using System.Text.Json; using AgentCore.Application.Transcript; using AgentCore.Infrastructure.Conversation.Postgres; using AgentCore.Infrastructure.Tests.Database.Postgres; @@ -7,7 +8,9 @@ namespace AgentCore.Infrastructure.Tests.Conversation.Postgres { /// - /// Retention, which deletes the conversation and lets the cascade take the words. + /// Retention, which deletes stale continuation rows only. A conversation and its words are + /// never swept: they stay whole for as long as the row lives, whether or not anything can + /// still resume it. /// public sealed class PostgresConversationStoreSweepTests : PostgresDatabaseTest { @@ -16,57 +19,52 @@ public sealed class PostgresConversationStoreSweepTests : PostgresDatabaseTest /// protected override bool Migrated => true; - private static ConversationMessageDraft Word() + private static JsonElement Envelope() { - return new(0, new ChatMessage(ChatRole.User, "hello"), "m0"); + using JsonDocument document = JsonDocument.Parse("""{ "state": {} }"""); + return document.RootElement.Clone(); } [PostgresFact] - public async Task SweepAsync_AConversationWhoseLastMessageIsOld_TakesTheConversationAndItsWords() + public async Task SweepAsync_AContinuationUntouchedPastTheWindow_TakesTheRow() { // Arrange PostgresConversationStore store = new(DataSource); - _ = await store.CreateAsync("old", Token); - _ = await store.AppendAsync("old", [Word()], cancellationToken: Token); - await ExecuteAsync("UPDATE agentcore.conversation_message SET updated_at = now() - interval '90 days'"); + _ = await store.CreateAsync("c1", Token); + await store.SaveContinuationAsync("resp_1", "c1", Envelope(), Token); + await ExecuteAsync("UPDATE agentcore.response_continuation SET updated_at = now() - interval '90 days'"); // Act int swept = await store.SweepAsync(Window, cancellationToken: Token); // Assert Assert.Equal(1, swept); - Assert.Equal(0L, await ScalarAsync("SELECT count(*) FROM agentcore.conversation")); - Assert.Equal(0L, await ScalarAsync("SELECT count(*) FROM agentcore.conversation_message")); + Assert.Null(await store.GetContinuationAsync("resp_1", Token)); } [PostgresFact] - public async Task SweepAsync_AConversationWithNoWordsAtAll_TakesItOnCreatedAt() + public async Task SweepAsync_AContinuationUntouchedPastTheWindow_LeavesTheConversationAndItsWords() { // Arrange PostgresConversationStore store = new(DataSource); - _ = await store.CreateAsync("empty", Token); - await ExecuteAsync("UPDATE agentcore.conversation SET created_at = now() - interval '90 days'"); + _ = await store.CreateAsync("c1", Token); + await store.SaveContinuationAsync("c1", "c1", Envelope(), Token); + await ExecuteAsync("UPDATE agentcore.response_continuation SET updated_at = now() - interval '90 days'"); // Act - int swept = await store.SweepAsync(Window, cancellationToken: Token); + _ = await store.SweepAsync(Window, cancellationToken: Token); // Assert - Assert.Equal(1, swept); - Assert.Equal(0L, await ScalarAsync("SELECT count(*) FROM agentcore.conversation")); + Assert.Equal(1L, await ScalarAsync("SELECT count(*) FROM agentcore.conversation")); } - /// - /// created_at is old and the message is fresh, so the message time must win. This is the whole - /// point of coalescing rather than reading either column alone. - /// [PostgresFact] - public async Task SweepAsync_AConversationSpokenOnInsideTheWindow_LeavesIt() + public async Task SweepAsync_AConversationRowItself_IsNeverTaken() { - // Arrange + // Arrange — an old conversation with no continuation at all: nothing here for the sweep to find. PostgresConversationStore store = new(DataSource); - _ = await store.CreateAsync("fresh", Token); - _ = await store.AppendAsync("fresh", [Word()], cancellationToken: Token); - await ExecuteAsync("UPDATE agentcore.conversation SET created_at = now() - interval '90 days'"); + _ = await store.CreateAsync("old", Token); + await ExecuteAsync("UPDATE agentcore.conversation SET created_at = now() - interval '400 days'"); // Act int swept = await store.SweepAsync(Window, cancellationToken: Token); @@ -77,43 +75,60 @@ public async Task SweepAsync_AConversationSpokenOnInsideTheWindow_LeavesIt() } [PostgresFact] - public async Task SweepAsync_ASweptConversation_LeavesItsAuditEvents() + public async Task SweepAsync_AContinuationWrittenInsideTheWindow_LeavesIt() { // Arrange PostgresConversationStore store = new(DataSource); - _ = await store.CreateAsync("old", Token); - await ExecuteAsync("UPDATE agentcore.conversation SET created_at = now() - interval '90 days'"); - await ExecuteAsync( - """ - INSERT INTO agentcore.audit_event (conversation_id, event_id, sequence, kind, occurred_at) - VALUES ('old', gen_random_uuid(), 0, 'conversation.started', now()) - """); + _ = await store.CreateAsync("c1", Token); + await store.SaveContinuationAsync("resp_1", "c1", Envelope(), Token); // Act - _ = await store.SweepAsync(Window, cancellationToken: Token); + int swept = await store.SweepAsync(Window, cancellationToken: Token); // Assert - Assert.Equal(1L, await ScalarAsync("SELECT count(*) FROM agentcore.audit_event")); + Assert.Equal(0, swept); + Assert.NotNull(await store.GetContinuationAsync("resp_1", Token)); } [PostgresFact] - public async Task SweepAsync_MoreExpiredConversationsThanOneBatch_LoopsUntilNoneAreLeft() + public async Task SweepAsync_ARewrittenContinuation_ResetsItsRetentionClock() + { + // Arrange — the same id written twice: the old timestamp must not survive the second write, + // or an actively resumed conversation would still age out from under it. + PostgresConversationStore store = new(DataSource); + _ = await store.CreateAsync("c1", Token); + await store.SaveContinuationAsync("c1", "c1", Envelope(), Token); + await ExecuteAsync("UPDATE agentcore.response_continuation SET updated_at = now() - interval '90 days'"); + await store.SaveContinuationAsync("c1", "c1", Envelope(), Token); + + // Act + int swept = await store.SweepAsync(Window, cancellationToken: Token); + + // Assert + Assert.Equal(0, swept); + Assert.NotNull(await store.GetContinuationAsync("c1", Token)); + } + + [PostgresFact] + public async Task SweepAsync_MoreExpiredContinuationsThanOneBatch_LoopsUntilNoneAreLeft() { // Arrange PostgresConversationStore store = new(DataSource); + _ = await store.CreateAsync("c1", Token); + for (int i = 0; i < 5; i++) { - _ = await store.CreateAsync($"c{i}", Token); + await store.SaveContinuationAsync($"resp_{i}", "c1", Envelope(), Token); } - await ExecuteAsync("UPDATE agentcore.conversation SET created_at = now() - interval '90 days'"); + await ExecuteAsync("UPDATE agentcore.response_continuation SET updated_at = now() - interval '90 days'"); // Act int swept = await store.SweepAsync(Window, batchSize: 2, cancellationToken: Token); // Assert Assert.Equal(5, swept); - Assert.Equal(0L, await ScalarAsync("SELECT count(*) FROM agentcore.conversation")); + Assert.Equal(0L, await ScalarAsync("SELECT count(*) FROM agentcore.response_continuation")); } [PostgresFact] @@ -122,7 +137,7 @@ public async Task EraseAsync_AConversationsWords_LeavesTheConversationItselfList // Arrange PostgresConversationStore store = new(DataSource); _ = await store.CreateAsync("c1", Token); - _ = await store.AppendAsync("c1", [Word()], cancellationToken: Token); + _ = await store.AppendAsync("c1", [new ConversationMessageDraft(0, new ChatMessage(ChatRole.User, "hello"), "m0")], cancellationToken: Token); // Act int erased = await store.EraseAsync("c1", Token); diff --git a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreTests.cs b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreTests.cs index d5f361d6..2fe40cf5 100644 --- a/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreTests.cs +++ b/tests/AgentCore.Infrastructure.Tests/Conversation/Postgres/PostgresConversationStoreTests.cs @@ -207,7 +207,7 @@ public async Task DeleteAsync_AConversation_TakesItsSameIdContinuationWithIt() PostgresConversationStore store = new(DataSource); _ = await store.CreateAsync("c1", Token); using JsonDocument document = JsonDocument.Parse("""{ "conversationId": "c1" }"""); - await store.SaveContinuationAsync("c1", document.RootElement, Token); + await store.SaveContinuationAsync("c1", "c1", document.RootElement, Token); // Act await store.DeleteAsync("c1", Token); diff --git a/tests/AgentCore.TestSupport/DelegatingConversationStore.cs b/tests/AgentCore.TestSupport/DelegatingConversationStore.cs index 6873f3e2..988a1ffa 100644 --- a/tests/AgentCore.TestSupport/DelegatingConversationStore.cs +++ b/tests/AgentCore.TestSupport/DelegatingConversationStore.cs @@ -149,9 +149,9 @@ public virtual ValueTask DetachPrincipalAsync( /// public virtual ValueTask SaveContinuationAsync( - string continuationId, JsonElement envelope, CancellationToken cancellationToken = default) + string continuationId, string conversationId, JsonElement envelope, CancellationToken cancellationToken = default) { - return Inner.SaveContinuationAsync(continuationId, envelope, cancellationToken); + return Inner.SaveContinuationAsync(continuationId, conversationId, envelope, cancellationToken); } ///