Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,26 @@ private static PartitionKey BuildPartitionKey(State state)
}

/// <inheritdoc />
protected override async ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default)
protected override ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default) =>
new(this.GetMessagesAsync(context.Session, cancellationToken));

/// <summary>
/// Gets the messages stored for the specified session.
/// </summary>
/// <param name="session">The agent session to get state from.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>
/// The messages in the timestamp-based order used by the agent invocation pipeline. Messages with equal stored
/// timestamps have no guaranteed relative order. When <see cref="MaxMessagesToRetrieve"/> is set, messages with
/// the latest timestamps are selected; if the limit intersects a timestamp tie, which tied messages are included
/// is unspecified.
/// </returns>
/// <remarks>
/// This method returns messages as stored and does not apply the output filter or chat-history source attribution
/// used by the agent invocation pipeline. Use <see cref="ChatHistoryProvider.InvokingAsync"/> when that processing
/// is required. <see cref="MaxItemCount"/> controls the query page size.
/// </remarks>
public async Task<IEnumerable<ChatMessage>> GetMessagesAsync(AgentSession? session, CancellationToken cancellationToken = default)
{
#pragma warning disable CA1513 // Use ObjectDisposedException.ThrowIf - not available on all target frameworks
if (this._disposed)
Expand All @@ -224,7 +243,7 @@ protected override async ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryA
}
#pragma warning restore CA1513

var state = this._sessionState.GetOrInitializeState(context.Session);
var state = this._sessionState.GetOrInitializeState(session);
var partitionKey = BuildPartitionKey(state);

// Fetch most recent messages in descending order when limit is set, then reverse to ascending
Expand All @@ -233,7 +252,7 @@ protected override async ValueTask<IEnumerable<ChatMessage>> ProvideChatHistoryA
.WithParameter("@conversationId", state.ConversationId)
.WithParameter("@type", "ChatMessage");
Comment thread
ilia-sokolov marked this conversation as resolved.

var iterator = this._container.GetItemQueryIterator<CosmosMessageDocument>(query, requestOptions: new QueryRequestOptions
using var iterator = this._container.GetItemQueryIterator<CosmosMessageDocument>(query, requestOptions: new QueryRequestOptions
{
PartitionKey = partitionKey,
MaxItemCount = this.MaxItemCount // Configurable query performance
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Azure.Core;
using Azure.Identity;
Expand Down Expand Up @@ -807,6 +808,137 @@ public async Task HierarchicalAndSimplePartitioning_ShouldCoexistAsync()
Assert.Equal("Hierarchical partitioning message", hierarchicalMessageList[0].Text);
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task GetMessagesAsync_WithMessages_ShouldReturnAllMessagesAcrossPagesAsync()
{
// Arrange
this.SkipIfEmulatorNotAvailable();
var session = CreateMockSession();
const string ConversationId = "get-messages-test";

using var provider = new CosmosChatHistoryProvider(this._connectionString, s_testDatabaseId, TestContainerId,
_ => new CosmosChatHistoryProvider.State(ConversationId))
{
MaxItemCount = 2,
};

List<ChatMessage> messages =
[
new(ChatRole.User, "Message 1"),
new(ChatRole.Assistant, "Message 2"),
new(ChatRole.User, "Message 3"),
new(ChatRole.Assistant, "Message 4"),
new(ChatRole.User, "Message 5"),
];

var context = new ChatHistoryProvider.InvokedContext(s_mockAgent, session, messages, []);
await provider.InvokedAsync(context);

// Wait a moment for eventual consistency
await Task.Delay(100);

// Act
var retrievedMessages = (await provider.GetMessagesAsync(session)).ToList();

// Assert
Assert.Equal(messages.Count, retrievedMessages.Count);
// Batch writes share a Unix-seconds timestamp, so this test intentionally verifies page completeness
// without asserting relative order among tied messages.
Assert.Equal(
messages.Select(message => message.Text).Order(),
retrievedMessages.Select(message => message.Text).Order());
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task GetMessagesAsync_WithNoMessages_ShouldReturnEmptyAsync()
{
// Arrange
this.SkipIfEmulatorNotAvailable();
var session = CreateMockSession();

using var provider = new CosmosChatHistoryProvider(this._connectionString, s_testDatabaseId, TestContainerId,
_ => new CosmosChatHistoryProvider.State("get-messages-empty-test"));

// Act
var messages = await provider.GetMessagesAsync(session);

// Assert
Assert.Empty(messages);
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task GetMessagesAsync_DoesNotApplyInvocationFilterOrSourceAttributionAsync()
{
// Arrange
this.SkipIfEmulatorNotAvailable();
var session = CreateMockSession();

using var provider = new CosmosChatHistoryProvider(
this._connectionString,
s_testDatabaseId,
TestContainerId,
_ => new CosmosChatHistoryProvider.State("get-messages-filter-test"),
provideOutputMessageFilter: messages => messages.Where(message => message.Text != "Hidden"));

List<ChatMessage> messages =
[
new(ChatRole.User, "Visible"),
new(ChatRole.Assistant, "Hidden"),
];

var invokedContext = new ChatHistoryProvider.InvokedContext(s_mockAgent, session, messages, []);
await provider.InvokedAsync(invokedContext);

// Act
var directMessages = (await provider.GetMessagesAsync(session)).ToList();
var invokingContext = new ChatHistoryProvider.InvokingContext(s_mockAgent, session, []);
var invocationMessages = (await provider.InvokingAsync(invokingContext)).ToList();

// Assert
Assert.Equal(2, directMessages.Count);
Assert.All(directMessages, message => Assert.Equal(AgentRequestMessageSourceType.External, message.GetAgentRequestMessageSourceType()));
Assert.Single(invocationMessages);
Assert.Equal("Visible", invocationMessages[0].Text);
Assert.Equal(AgentRequestMessageSourceType.ChatHistory, invocationMessages[0].GetAgentRequestMessageSourceType());
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task GetMessagesAsync_AfterDispose_ShouldThrowObjectDisposedExceptionAsync()
{
// Arrange
this.SkipIfEmulatorNotAvailable();
var session = CreateMockSession();
var provider = new CosmosChatHistoryProvider(this._connectionString, s_testDatabaseId, TestContainerId,
_ => new CosmosChatHistoryProvider.State("get-messages-disposed-test"));
provider.Dispose();

// Act & Assert
await Assert.ThrowsAsync<ObjectDisposedException>(() => provider.GetMessagesAsync(session));
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task GetMessagesAsync_WithCanceledToken_ShouldThrowOperationCanceledExceptionAsync()
{
// Arrange
this.SkipIfEmulatorNotAvailable();
var session = CreateMockSession();

using var provider = new CosmosChatHistoryProvider(this._connectionString, s_testDatabaseId, TestContainerId,
_ => new CosmosChatHistoryProvider.State("get-messages-cancellation-test"));

using var cancellationTokenSource = new CancellationTokenSource();
cancellationTokenSource.Cancel();

// Act & Assert
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => provider.GetMessagesAsync(session, cancellationTokenSource.Token));
}

[Fact]
[Trait("Category", "CosmosDB")]
public async Task MaxMessagesToRetrieve_ShouldLimitAndReturnMostRecentAsync()
Expand Down
Loading