.Net: Processes: Checkpoint state improvements (#12759)

### Description

- Checkpoint improvements that allow interacting with storage in
particular times -> end of superstep, initialization only.
- Checkpoint allows "staging" states from process and steps and only
uploading to cloud at the end of superstep
- Separating functionality available for process vs step
- Adding/Updating tests
- Adding simple cosmos db connector for LocalRuntime demo

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [ ] The code builds clean without any errors or warnings
- [ ] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [ ] All unit tests pass, and I have added new tests where possible
- [ ] I didn't break anyone 😄

---------

Co-authored-by: Estefania Tenorio <[email protected]>

Author: esttenorio

dotnet/SK-dotnet.slnx

@@ -79,7 +79,7 @@
   </Folder>
   <Folder Name="/samples/Demos/ProcessWithCloudEvents/">
     <File Path="samples/Demos/ProcessWithCloudEvents/README.md" />
-    <Project Path="samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc/ProcessWithCloudEvents.Grpc.csproj" />
+    <Project Path="samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc.LocalRuntime/ProcessWithCloudEvents.Grpc.LocalRuntime.csproj" />
     <Project Path="samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Processes/ProcessWithCloudEvents.Processes.csproj" />
   </Folder>
   <Folder Name="/src/">

dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc.LocalRuntime/ProcessWithCloudEvents.Grpc.LocalRuntime.csproj

@@ -37,6 +37,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+    <PackageReference Include="Microsoft.Azure.Cosmos" />
   </ItemGroup>
 
 </Project>

dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc.LocalRuntime/Program.cs

@@ -27,6 +27,7 @@
 });
 
 var openAIOptions = config.GetValid<OpenAIOptions>(OpenAIOptions.SectionName);
+var cosmosDbOptions = config.GetValid<CosmosDBOptions>(CosmosDBOptions.SectionName);
 
 // Configure the Kernel with DI. This is required for dependency injection to work with processes.
 builder.Services.AddKernel();
@@ -65,7 +66,13 @@
 // Registering storage used for persisting process state with Local Runtime
 string tempDirectoryPath = Path.Combine(Path.GetTempPath(), "MySKProcessStorage");
 var storageInstance = new JsonFileStorage(tempDirectoryPath);
-builder.Services.AddSingleton<IProcessStorageConnector>(storageInstance);
+
+var cloudStorageInstance = new CosmosDbProcessStorageConnector(
+    cosmosDbOptions.ConnectionString, cosmosDbOptions.DatabaseName, cosmosDbOptions.ContainerName
+);
+
+//builder.Services.AddSingleton<IProcessStorageConnector>(storageInstance);
+builder.Services.AddSingleton<IProcessStorageConnector>(cloudStorageInstance);
 
 // Enabling CORS for grpc-web when using webApp as client, remove if not needed
 builder.Services.AddCors(o => o.AddPolicy("AllowAll", builder =>

dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Utilities/Options/CosmosDBOptions.cs

@@ -0,0 +1,22 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.ComponentModel.DataAnnotations;
+
+namespace ProcessWithCloudEvents.SharedComponents.Options;
+
+/// <summary>
+/// Configuration for Cosmos DB.
+/// </summary>
+public class CosmosDBOptions
+{
+    public const string SectionName = "CosmosDB";
+
+    [Required]
+    public string ConnectionString { get; set; } = string.Empty;
+
+    [Required]
+    public string DatabaseName { get; set; } = string.Empty;
+
+    [Required]
+    public string ContainerName { get; set; } = string.Empty;
+}

dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Utilities/Storage/CosmosDbContainerStorageConnector.cs

@@ -0,0 +1,108 @@
+// Copyright (c) Microsoft. All rights reserved.
+#pragma warning disable IDE0005 // Using directive is unnecessary
+using System.Collections.Concurrent;
+using System.ComponentModel;
+using System.Text.Json;
+using Microsoft.Azure.Cosmos;
+using Microsoft.SemanticKernel;
+using Newtonsoft.Json;
+#pragma warning restore IDE0005 // Using directive is unnecessary
+
+namespace ProcessWithCloudEvents.SharedComponents.Storage;
+
+// CosmosDB V3 has a dependency on Newtonsoft.Json, so need to add wrapper class for Cosmos DB entities:
+// https://brettmckenzie.net/posts/the-input-content-is-invalid-because-the-required-properties-id-are-missing/
+internal record CosmosDbEntity<T>
+{
+    [JsonProperty("id")]
+    public string Id { get; init; } = string.Empty;
+
+    [JsonProperty("body")]
+    public T Body { get; init; } = default!;
+
+    [JsonProperty("instanceId")]
+    public string PartitionKey { get; init; } = string.Empty;
+}
+
+internal sealed class CosmosDbProcessStorageConnector : IProcessStorageConnector, IDisposable
+{
+    private readonly CosmosClient _cosmosClient;
+    private readonly Microsoft.Azure.Cosmos.Container _container;
+    private readonly string _databaseId;
+    private readonly string _containerId;
+
+    public CosmosDbProcessStorageConnector(string connectionString, string databaseId, string containerId)
+    {
+        this._cosmosClient = new CosmosClient(connectionString);
+        this._databaseId = databaseId;
+        this._containerId = containerId;
+        this._container = this._cosmosClient.GetContainer(databaseId, containerId);
+    }
+
+    public async ValueTask OpenConnectionAsync()
+    {
+        // Cosmos DB client handles connection pooling internally, so just ensure the client is initialized
+        await Task.CompletedTask;
+    }
+
+    public async ValueTask CloseConnectionAsync()
+    {
+        // Dispose the CosmosClient to close connections  
+        this._cosmosClient.Dispose();
+        await Task.CompletedTask;
+    }
+
+    public async Task<TEntry?> GetEntryAsync<TEntry>(string id) where TEntry : class
+    {
+        try
+        {
+            var response = await this._container.ReadItemAsync<CosmosDbEntity<TEntry>>(id, new PartitionKey(id));
+            return response.Resource.Body;
+        }
+        catch (CosmosException ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound)
+        {
+            // Item not found  
+            return null;
+        }
+    }
+
+    public async Task<bool> SaveEntryAsync<TEntry>(string id, string type, TEntry entry) where TEntry : class
+    {
+        try
+        {
+            var wrappedEntry = new CosmosDbEntity<TEntry>
+            {
+                Id = id,
+                Body = entry,
+                PartitionKey = id
+            };
+            await this._container.UpsertItemAsync(wrappedEntry, new PartitionKey(id));
+            return true;
+        }
+        catch (CosmosException ex)
+        {
+            // Handle exceptions as needed, log them, etc.  
+            Console.WriteLine($"Error saving entry: {ex.Message}");
+            return false;
+        }
+    }
+
+    public async Task<bool> DeleteEntryAsync(string id)
+    {
+        try
+        {
+            await this._container.DeleteItemAsync<object>(id, new PartitionKey(id));
+            return true;
+        }
+        catch (CosmosException ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound)
+        {
+            // Item not found  
+            return false;
+        }
+    }
+
+    public void Dispose()
+    {
+        this._cosmosClient?.Dispose();
+    }
+}

dotnet/src/Experimental/Process.Abstractions/IProcessStepStorageOperations.cs

@@ -0,0 +1,64 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.SemanticKernel.Process.Models.Storage;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// Defines operations for managing the storage of process step information, state, and events.
+/// </summary>
+public interface IProcessStepStorageOperations
+{
+    // For now step data is fetched by parent process only.
+    // Task FetchStepDataAsync(KernelProcessStepInfo step);
+
+    /// <summary>
+    /// Retrieves detailed information about a specific process step: parentId, version, etc.
+    /// </summary>
+    /// <param name="stepInfo">The step for which information is to be retrieved. This parameter cannot be null.</param>
+    /// <returns>A task that represents the asynchronous operation. The task result contains a <see cref="StorageStepInfo"/>
+    /// object with the step details, or <see langword="null"/> if the step information is not available.</returns>
+    Task<StorageStepInfo?> GetStepInfoAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Saves detailed information about a specific process step, such as parentId, version, etc.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns></returns>
+    Task<bool> SaveStepInfoAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Retrieves the current state of a process step.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns>A task that represents the asynchronous operation. The task result contains the current state of the specified
+    /// step, or <see langword="null"/> if the step does not exist.</returns>
+    Task<KernelProcessStepState?> GetStepStateAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Saves the current state of a process step.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns></returns>
+    Task<bool> SaveStepStateAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Retrieves the events associated with a specific process step.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns></returns>
+    Task<StorageStepEvents?> GetStepEventsAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Saves the events associated with a specific process step.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <param name="edgeGroups"></param>
+    /// <returns></returns>
+    Task<bool> SaveStepEventsAsync(KernelProcessStepInfo stepInfo, Dictionary<string, Dictionary<string, object?>>? edgeGroups = null);
+    // For now step data can be saved to storage only by parent process only.
+    // This is to only save data to storage at the end of the process super step execution.
+    //Task<bool> SaveStepDataToStorageAsync(KernelProcessStepInfo step);
+}

dotnet/src/Experimental/Process.Abstractions/IProcessStorageOperations.cs

@@ -0,0 +1,67 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.SemanticKernel.Process.Models.Storage;
+
+namespace Microsoft.SemanticKernel;
+
+internal interface IProcessStorageOperations
+{
+    /// <summary>
+    /// Initializes the storage connection to be used by the process.
+    /// </summary>
+    /// <returns></returns>
+    Task<bool> InitializeAsync();
+    /// <summary>
+    /// Closes storage connection and cleans up resources.
+    /// </summary>
+    /// <returns></returns>
+    Task<bool> CloseAsync();
+    /// <summary>
+    /// Fetches from storage the process data for a specific process.
+    /// </summary>
+    /// <param name="process"></param>
+    /// <returns></returns>
+    Task FetchProcessDataAsync(KernelProcess process);
+    /// <summary>
+    /// Get the process information already retrieved from storage, including parentId, version, mapping of steps and running ids.
+    /// </summary>
+    /// <param name="process"></param>
+    /// <returns></returns>
+    Task<StorageProcessInfo?> GetProcessInfoAsync(KernelProcess process);
+    /// <summary>
+    /// Saves the process information to storage, including parentId, version, mapping of steps and running ids.
+    /// </summary>
+    /// <param name="process"></param>
+    /// <returns></returns>
+    Task<bool> SaveProcessInfoAsync(KernelProcess process);
+    /// <summary>
+    /// Save process events to storage, including pending external events.
+    /// </summary>
+    /// <param name="process"></param>
+    /// <param name="pendingExternalEvents"></param>
+    /// <returns></returns>
+    Task<bool> SaveProcessEventsAsync(KernelProcess process, List<KernelProcessEvent>? pendingExternalEvents = null);
+    /// <summary>
+    /// Saves all process related data to storage, including process info, events, and step data.
+    /// </summary>
+    /// <param name="process"></param>
+    /// <returns></returns>
+    Task<bool> SaveProcessDataToStorageAsync(KernelProcess process);
+
+    // Step related operations to be applied to process children steps only
+    /// <summary>
+    /// Fetches from storage the step data for a specific process step.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns></returns>
+    Task FetchStepDataAsync(KernelProcessStepInfo stepInfo);
+
+    /// <summary>
+    /// Saves the step data to storage.
+    /// </summary>
+    /// <param name="stepInfo"></param>
+    /// <returns></returns>
+    Task<bool> SaveStepDataToStorageAsync(KernelProcessStepInfo stepInfo);
+}

dotnet/src/Experimental/Process.Abstractions/KernelProcessStepInfo.cs

@@ -31,6 +31,15 @@ public KernelProcessStepState State
         }
     }
 
+    /// <inheritdoc cref="KernelProcessStepState.RunId"/>
+    public string? RunId => this.State.RunId;
+
+    /// <inheritdoc cref="KernelProcessStepState.StepId"/>
+    public string? StepId => this.State.StepId;
+
+    /// <inheritdoc cref="KernelProcessStepState.ParentId"/>
+    public string? ParentId => this.State.ParentId;
+
     /// <summary>
     /// The semantic description of the Step. This is intended to be human and AI readable and is not required to be unique.
     /// </summary>

dotnet/src/Experimental/Process.Abstractions/KernelProcessStepState.cs

@@ -44,6 +44,13 @@ internal static void RegisterDerivedType(Type derivedType)
     [JsonPropertyName("runId")]
     public string? RunId { get; set; }
 
+    /// <summary>
+    /// Gets or sets the identifier of the step parent.
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("parentId")]
+    public string? ParentId { get; set; }
+
     /// <summary>
     /// The name of the Step. This is intended to be human readable and is not required to be unique. If
     /// not provided, the name will be derived from the steps .NET type.
@@ -105,4 +112,13 @@ public KernelProcessStepState(string stepId, string version, string? runId = nul
         this.StepId = stepId;
         this.Version = version;
     }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="KernelProcessStepState"/> class.
+    /// </summary>
+    /// <param name="stepState"></param>
+    public KernelProcessStepState(KernelProcessStepState stepState)
+        : base(stepState.StepId, stepState.Version, stepState.RunId)
+    {
+    }
 }

dotnet/src/Experimental/Process.Abstractions/Models/Storage/StorageEntryBase.cs

@@ -0,0 +1,17 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Text.Json.Serialization;
+
+namespace Microsoft.SemanticKernel.Process.Models.Storage;
+
+/// <summary>
+/// Base class for storage entries.
+/// </summary>
+public abstract record StorageEntryBase
+{
+    /// <summary>
+    /// Unique identifier of the storage entry.
+    /// </summary>
+    [JsonPropertyName("instanceId")]
+    public string InstanceId { get; set; } = string.Empty;
+}