// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using Azure.Core;
using Microsoft.Azure.Cosmos;
using Microsoft.Shared.Diagnostics;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
namespace Microsoft.Agents.AI.Workflows.Checkpointing;
///
/// Provides a Cosmos DB implementation of the abstract class.
///
/// The type of objects to store as checkpoint values.
[RequiresUnreferencedCode("The CosmosCheckpointStore uses JSON serialization which is incompatible with trimming.")]
[RequiresDynamicCode("The CosmosCheckpointStore uses JSON serialization which is incompatible with NativeAOT.")]
public class CosmosCheckpointStore : JsonCheckpointStore, IDisposable
{
private readonly CosmosClient _cosmosClient;
private readonly Container _container;
private readonly bool _ownsClient;
private bool _disposed;
///
/// Initializes a new instance of the class using a connection string.
///
/// The Cosmos DB connection string.
/// The identifier of the Cosmos DB database.
/// The identifier of the Cosmos DB container.
/// Thrown when any required parameter is null.
/// Thrown when any string parameter is null or whitespace.
public CosmosCheckpointStore(string connectionString, string databaseId, string containerId)
{
var cosmosClientOptions = new CosmosClientOptions();
this._cosmosClient = new CosmosClient(Throw.IfNullOrWhitespace(connectionString), cosmosClientOptions);
this._container = this._cosmosClient.GetContainer(Throw.IfNullOrWhitespace(databaseId), Throw.IfNullOrWhitespace(containerId));
this._ownsClient = true;
}
///
/// Initializes a new instance of the class using a TokenCredential for authentication.
///
/// The Cosmos DB account endpoint URI.
/// The TokenCredential to use for authentication (e.g., DefaultAzureCredential, ManagedIdentityCredential).
/// The identifier of the Cosmos DB database.
/// The identifier of the Cosmos DB container.
/// Thrown when any required parameter is null.
/// Thrown when any string parameter is null or whitespace.
public CosmosCheckpointStore(string accountEndpoint, TokenCredential tokenCredential, string databaseId, string containerId)
{
var cosmosClientOptions = new CosmosClientOptions
{
SerializerOptions = new CosmosSerializationOptions
{
PropertyNamingPolicy = CosmosPropertyNamingPolicy.CamelCase
}
};
this._cosmosClient = new CosmosClient(Throw.IfNullOrWhitespace(accountEndpoint), Throw.IfNull(tokenCredential), cosmosClientOptions);
this._container = this._cosmosClient.GetContainer(Throw.IfNullOrWhitespace(databaseId), Throw.IfNullOrWhitespace(containerId));
this._ownsClient = true;
}
///
/// Initializes a new instance of the class using an existing .
///
/// The instance to use for Cosmos DB operations.
/// The identifier of the Cosmos DB database.
/// The identifier of the Cosmos DB container.
/// Thrown when is null.
/// Thrown when any string parameter is null or whitespace.
public CosmosCheckpointStore(CosmosClient cosmosClient, string databaseId, string containerId)
{
this._cosmosClient = Throw.IfNull(cosmosClient);
this._container = this._cosmosClient.GetContainer(Throw.IfNullOrWhitespace(databaseId), Throw.IfNullOrWhitespace(containerId));
this._ownsClient = false;
}
///
/// Gets the identifier of the Cosmos DB database.
///
public string DatabaseId => this._container.Database.Id;
///
/// Gets the identifier of the Cosmos DB container.
///
public string ContainerId => this._container.Id;
///
public override async ValueTask CreateCheckpointAsync(string runId, JsonElement value, CheckpointInfo? parent = null)
{
if (string.IsNullOrWhiteSpace(runId))
{
throw new ArgumentException("Cannot be null or whitespace", nameof(runId));
}
#pragma warning disable CA1513 // Use ObjectDisposedException.ThrowIf - not available on all target frameworks
if (this._disposed)
{
throw new ObjectDisposedException(this.GetType().FullName);
}
#pragma warning restore CA1513
var checkpointId = Guid.NewGuid().ToString("N");
var checkpointInfo = new CheckpointInfo(runId, checkpointId);
var document = new CosmosCheckpointDocument
{
Id = $"{runId}_{checkpointId}",
RunId = runId,
CheckpointId = checkpointId,
Value = JToken.Parse(value.GetRawText()),
ParentCheckpointId = parent?.CheckpointId,
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds()
};
await this._container.CreateItemAsync(document, new PartitionKey(runId)).ConfigureAwait(false);
return checkpointInfo;
}
///
public override async ValueTask RetrieveCheckpointAsync(string runId, CheckpointInfo key)
{
if (string.IsNullOrWhiteSpace(runId))
{
throw new ArgumentException("Cannot be null or whitespace", nameof(runId));
}
if (key is null)
{
throw new ArgumentNullException(nameof(key));
}
#pragma warning disable CA1513 // Use ObjectDisposedException.ThrowIf - not available on all target frameworks
if (this._disposed)
{
throw new ObjectDisposedException(this.GetType().FullName);
}
#pragma warning restore CA1513
var id = $"{runId}_{key.CheckpointId}";
try
{
var response = await this._container.ReadItemAsync(id, new PartitionKey(runId)).ConfigureAwait(false);
using var document = JsonDocument.Parse(response.Resource.Value.ToString());
return document.RootElement.Clone();
}
catch (CosmosException ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound)
{
throw new InvalidOperationException($"Checkpoint with ID '{key.CheckpointId}' for run '{runId}' not found.");
}
}
///
public override async ValueTask> RetrieveIndexAsync(string runId, CheckpointInfo? withParent = null)
{
if (string.IsNullOrWhiteSpace(runId))
{
throw new ArgumentException("Cannot be null or whitespace", nameof(runId));
}
#pragma warning disable CA1513 // Use ObjectDisposedException.ThrowIf - not available on all target frameworks
if (this._disposed)
{
throw new ObjectDisposedException(this.GetType().FullName);
}
#pragma warning restore CA1513
QueryDefinition query = withParent == null
? new QueryDefinition("SELECT c.runId, c.checkpointId FROM c WHERE c.runId = @runId ORDER BY c.timestamp ASC")
.WithParameter("@runId", runId)
: new QueryDefinition("SELECT c.runId, c.checkpointId FROM c WHERE c.runId = @runId AND c.parentCheckpointId = @parentCheckpointId ORDER BY c.timestamp ASC")
.WithParameter("@runId", runId)
.WithParameter("@parentCheckpointId", withParent.CheckpointId);
var iterator = this._container.GetItemQueryIterator(query);
var checkpoints = new List();
while (iterator.HasMoreResults)
{
var response = await iterator.ReadNextAsync().ConfigureAwait(false);
checkpoints.AddRange(response.Select(r => new CheckpointInfo(r.RunId, r.CheckpointId)));
}
return checkpoints;
}
///
public void Dispose()
{
this.Dispose(true);
GC.SuppressFinalize(this);
}
///
/// Releases the unmanaged resources used by the and optionally releases the managed resources.
///
/// true to release both managed and unmanaged resources; false to release only unmanaged resources.
protected virtual void Dispose(bool disposing)
{
if (!this._disposed)
{
if (disposing && this._ownsClient)
{
this._cosmosClient?.Dispose();
}
this._disposed = true;
}
}
/// Represents a checkpoint document stored in Cosmos DB.
internal sealed class CosmosCheckpointDocument
{
[JsonProperty("id")]
public string Id { get; set; } = string.Empty;
[JsonProperty("runId")]
public string RunId { get; set; } = string.Empty;
[JsonProperty("checkpointId")]
public string CheckpointId { get; set; } = string.Empty;
[JsonProperty("value")]
public JToken Value { get; set; } = JValue.CreateNull();
[JsonProperty("parentCheckpointId")]
public string? ParentCheckpointId { get; set; }
[JsonProperty("timestamp")]
public long Timestamp { get; set; }
}
///
/// Represents the result of a checkpoint query.
///
[SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by Cosmos DB query deserialization")]
private sealed class CheckpointQueryResult
{
public string RunId { get; set; } = string.Empty;
public string CheckpointId { get; set; } = string.Empty;
}
}
///
/// Provides a non-generic Cosmos DB implementation of the abstract class.
///
[RequiresUnreferencedCode("The CosmosCheckpointStore uses JSON serialization which is incompatible with trimming.")]
[RequiresDynamicCode("The CosmosCheckpointStore uses JSON serialization which is incompatible with NativeAOT.")]
public sealed class CosmosCheckpointStore : CosmosCheckpointStore
{
///
public CosmosCheckpointStore(string connectionString, string databaseId, string containerId)
: base(connectionString, databaseId, containerId)
{
}
///
public CosmosCheckpointStore(string accountEndpoint, TokenCredential tokenCredential, string databaseId, string containerId)
: base(accountEndpoint, tokenCredential, databaseId, containerId)
{
}
///
public CosmosCheckpointStore(CosmosClient cosmosClient, string databaseId, string containerId)
: base(cosmosClient, databaseId, containerId)
{
}
}