.Net: Processes: Hook up uninitialization for Steps in Local Runtime

dotnet/samples/Demos/ProcessFrameworkWithAspire/ProcessFramework.Aspire/ProcessFramework.Aspire.ProcessOrchestrator/Program.cs

@@ -77,7 +77,7 @@
         .StopProcess();
 
     var process = processBuilder.Build();
-    using var runningProcess = await process.StartAsync(
+    await using var runningProcess = await process.StartAsync(
         kernel,
         new KernelProcessEvent { Id = ProcessEvents.TranslateDocument, Data = "COME I FORNITORI INFLUENZANO I TUOI COSTI Quando scegli un piano di assicurazione sanitaria, uno dei fattori più importanti da considerare è la rete di fornitori in convenzione disponibili con il piano. Northwind Standard offre un'ampia varietà di fornitori in convenzione, tra cui medici di base, specialisti, ospedali e farmacie. Questo ti permette di scegliere un fornitore comodo per te e la tua famiglia, contribuendo al contempo a mantenere bassi i tuoi costi. Se scegli un fornitore in convenzione con il tuo piano, pagherai generalmente copay e franchigie più basse rispetto a un fornitore fuori rete. Inoltre, molti servizi, come l'assistenza preventiva, possono essere coperti senza alcun costo aggiuntivo se ricevuti da un fornitore in convenzione. È importante notare, tuttavia, che Northwind Standard non copre i servizi di emergenza, l'assistenza per la salute mentale e l'abuso di sostanze, né i servizi fuori rete. Questo significa che potresti dover pagare di tasca tua per questi servizi se ricevuti da un fornitore fuori rete. Quando scegli un fornitore in convenzione, ci sono alcuni suggerimenti da tenere a mente. Verifica che il fornitore sia in convenzione con il tuo piano. Puoi confermarlo chiamando l'ufficio del fornitore e chiedendo se è in rete con Northwind Standard. Puoi anche utilizzare lo strumento di ricerca fornitori sul sito web di Northwind Health per verificare la copertura. Assicurati che il fornitore stia accettando nuovi pazienti. Alcuni fornitori potrebbero essere in convenzione ma non accettare nuovi pazienti. Considera la posizione del fornitore. Se il fornitore è troppo lontano, potrebbe essere difficile raggiungere gli appuntamenti. Valuta gli orari dell'ufficio del fornitore. Se lavori durante il giorno, potresti aver bisogno di trovare un fornitore con orari serali o nel fine settimana. Scegliere un fornitore in convenzione può aiutarti a risparmiare sui costi sanitari. Seguendo i suggerimenti sopra e facendo ricerche sulle opzioni disponibili, puoi trovare un fornitore conveniente, accessibile e in rete con il tuo piano Northwind Standard." }
     );

dotnet/samples/GettingStartedWithProcesses/Step00/Step00_Processes.cs

@@ -59,7 +59,7 @@ public async Task UseSimplestProcessAsync()
         KernelProcess kernelProcess = process.Build();
 
         // Start the process with an initial external event
-        using var runningProcess = await kernelProcess.StartAsync(
+        await using var runningProcess = await kernelProcess.StartAsync(
             kernel,
                 new KernelProcessEvent()
                 {

dotnet/samples/GettingStartedWithProcesses/Step01/Step01_Processes.cs

@@ -77,7 +77,7 @@ public async Task UseSimpleProcessAsync()
         Console.WriteLine($"Diagram generated at: {generatedImagePath}");
 
         // Start the process with an initial external event
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = ChatBotEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = ChatBotEvents.StartProcess, Data = null });
     }
 
     /// <summary>

dotnet/samples/GettingStartedWithProcesses/Step02/Step02a_AccountOpening.cs

@@ -145,7 +145,7 @@ public async Task UseAccountOpeningProcessSuccessfulInteractionAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputSuccessfulInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 
     /// <summary>
@@ -156,7 +156,7 @@ public async Task UseAccountOpeningProcessFailureDueToCreditScoreFailureAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputCreditScoreFailureInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 
     /// <summary>
@@ -167,6 +167,6 @@ public async Task UseAccountOpeningProcessFailureDueToFraudFailureAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputFraudFailureInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 }

dotnet/samples/GettingStartedWithProcesses/Step02/Step02b_AccountOpening.cs

@@ -112,7 +112,7 @@ public async Task UseAccountOpeningProcessSuccessfulInteractionAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputSuccessfulInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 
     /// <summary>
@@ -123,7 +123,7 @@ public async Task UseAccountOpeningProcessFailureDueToCreditScoreFailureAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputCreditScoreFailureInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 
     /// <summary>
@@ -134,6 +134,6 @@ public async Task UseAccountOpeningProcessFailureDueToFraudFailureAsync()
     {
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SetupAccountOpeningProcess<UserInputFraudFailureInteractionStep>();
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent() { Id = AccountOpeningEvents.StartProcess, Data = null });
     }
 }

dotnet/samples/GettingStartedWithProcesses/Step03/Step03a_FoodPreparation.cs

@@ -252,7 +252,7 @@ protected async Task UsePrepareSpecificProductAsync(ProcessBuilder processBuilde
 
         // Assert
         Console.WriteLine($"=== Start SK Process '{processBuilder.Name}' ===");
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent()
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent()
         {
             Id = externalTriggerEvent, Data = new List<string>()
         });

dotnet/samples/GettingStartedWithProcesses/Step03/Step03b_FoodOrdering.cs

@@ -45,7 +45,7 @@ protected async Task UsePrepareFoodOrderProcessSingleItemAsync(FoodItem foodItem
         Kernel kernel = CreateKernelWithChatCompletion();
         KernelProcess kernelProcess = SingleFoodItemProcess.CreateProcess().Build();
 
-        using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent()
+        await using var runningProcess = await kernelProcess.StartAsync(kernel, new KernelProcessEvent()
         {
             Id = SingleFoodItemProcess.ProcessEvents.SingleOrderReceived,
             Data = foodItem

dotnet/samples/GettingStartedWithProcesses/Step04/Step04_AgentOrchestration.cs

@@ -65,7 +65,7 @@ private async Task RunProcessAsync(KernelProcess process)
         Kernel kernel = SetupKernel(history);
 
         // Execute process
-        using LocalKernelProcessContext localProcess =
+        await using LocalKernelProcessContext localProcess =
             await process.StartAsync(
                 kernel,
                 new KernelProcessEvent()

dotnet/samples/GettingStartedWithProcesses/Step05/Step05_MapReduce.cs

@@ -44,7 +44,7 @@ public async Task RunMapReduceAsync()
 
         // Execute the process
         Kernel kernel = new();
-        using LocalKernelProcessContext localProcess =
+        await using LocalKernelProcessContext localProcess =
             await process.StartAsync(
                 kernel,
                 new KernelProcessEvent

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

@@ -10,17 +10,14 @@ namespace Microsoft.SemanticKernel;
 public sealed class KernelProcessStepContext
 {
     private readonly IKernelProcessMessageChannel _stepMessageChannel;
-    private readonly IExternalKernelProcessMessageChannel? _externalMessageChannel;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="KernelProcessStepContext"/> class.
     /// </summary>
     /// <param name="channel">An instance of <see cref="IKernelProcessMessageChannel"/>.</param>
-    /// <param name="externalMessageChannel">An instance of <see cref="IExternalKernelProcessMessageChannel"/></param>
-    public KernelProcessStepContext(IKernelProcessMessageChannel channel, IExternalKernelProcessMessageChannel? externalMessageChannel = null)
+    public KernelProcessStepContext(IKernelProcessMessageChannel channel)
     {
         this._stepMessageChannel = channel;
-        this._externalMessageChannel = externalMessageChannel;
     }
 
     /// <summary>
@@ -55,21 +52,4 @@ public ValueTask EmitEventAsync(
                 Visibility = visibility
             });
     }
-
-    /// <summary>
-    /// Emit an external event to through a <see cref="IExternalKernelProcessMessageChannel"/>
-    /// component if connected from within the SK process
-    /// </summary>
-    /// <param name="processEventData">data containing event details</param>
-    /// <returns></returns>
-    /// <exception cref="KernelException"></exception>
-    public async Task EmitExternalEventAsync(KernelProcessProxyMessage processEventData)
-    {
-        if (this._externalMessageChannel == null)
-        {
-            throw new KernelException($"External message channel not configured for step with topic {processEventData.ExternalTopicName}");
-        }
-
-        await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, processEventData).ConfigureAwait(false);
-    }
 }

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

@@ -0,0 +1,54 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Threading.Tasks;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// Provides step related functionality for Kernel Functions running in a step to emit events externally.
+/// </summary>
+public class KernelProcessStepExternalContext
+{
+    private readonly IExternalKernelProcessMessageChannel? _externalMessageChannel;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="KernelProcessStepContext"/> class.
+    /// </summary>
+    /// <param name="externalMessageChannel">An instance of <see cref="IExternalKernelProcessMessageChannel"/></param>
+    public KernelProcessStepExternalContext(IExternalKernelProcessMessageChannel? externalMessageChannel = null)
+    {
+        this._externalMessageChannel = externalMessageChannel;
+    }
+
+    /// <summary>
+    /// Emit an external event to through a <see cref="IExternalKernelProcessMessageChannel"/>
+    /// component if connected from within the SK process
+    /// </summary>
+    /// <param name="processEventData">data containing event details</param>
+    /// <returns></returns>
+    /// <exception cref="KernelException"></exception>
+    public async Task EmitExternalEventAsync(KernelProcessProxyMessage processEventData)
+    {
+        if (this._externalMessageChannel == null)
+        {
+            throw new KernelException($"External message channel not configured for step with topic {processEventData.ExternalTopicName}");
+        }
+
+        await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, processEventData).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Closes connection with external messaging channel
+    /// </summary>
+    /// <returns><see cref="Task"/></returns>
+    /// <exception cref="KernelException"></exception>
+    public async Task CloseExternalEventChannelAsync()
+    {
+        if (this._externalMessageChannel == null)
+        {
+            throw new KernelException("External message channel not configured for step");
+        }
+
+        await this._externalMessageChannel.Uninitialize().ConfigureAwait(false);
+    }
+}

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

@@ -20,14 +20,24 @@ public static class Functions
         public const string EmitExternalEvent = nameof(EmitExternalEvent);
     }
 
+    /// <summary>
+    /// On deactivation, external communication channel must be closed
+    /// </summary>
+    /// <param name="context">instance of <see cref="KernelProcessStepContext"/></param>
+    /// <returns></returns>
+    public async ValueTask DeactivateAsync(KernelProcessStepExternalContext context)
+    {
+        await context.CloseExternalEventChannelAsync().ConfigureAwait(false);
+    }
+
     /// <summary>
     /// Step function used to emit events externally
     /// </summary>
     /// <param name="context">instance of <see cref="KernelProcessStepContext"/></param>
     /// <param name="proxyEvent">event data passed to proxy step</param>
     /// <returns></returns>
     [KernelFunction(Functions.EmitExternalEvent)]
-    public Task EmitExternalEventAsync(KernelProcessStepContext context, KernelProcessProxyMessage proxyEvent)
+    public Task EmitExternalEventAsync(KernelProcessStepExternalContext context, KernelProcessProxyMessage proxyEvent)
     {
         Verify.NotNull(proxyEvent.ExternalTopicName, nameof(proxyEvent.ExternalTopicName));
         return context.EmitExternalEventAsync(proxyEvent);

dotnet/src/Experimental/Process.LocalRuntime/LocalKernelProcessContext.cs

@@ -1,5 +1,4 @@
 // Copyright (c) Microsoft. All rights reserved.
-using System;
 using System.Threading.Tasks;
 using Microsoft.SemanticKernel.Process;
 
@@ -8,7 +7,7 @@ namespace Microsoft.SemanticKernel;
 /// <summary>
 /// Provides context and actions on a process that is running locally.
 /// </summary>
-public sealed class LocalKernelProcessContext : KernelProcessContext, IDisposable
+public sealed class LocalKernelProcessContext : KernelProcessContext, System.IAsyncDisposable
 {
     private readonly LocalProcess _localProcess;
     private readonly Kernel _kernel;
@@ -53,7 +52,10 @@ public override Task SendEventAsync(KernelProcessEvent processEvent) =>
     /// <summary>
     /// Disposes of the resources used by the process.
     /// </summary>
-    public void Dispose() => this._localProcess.Dispose();
+    public async ValueTask DisposeAsync()
+    {
+        await this._localProcess.DisposeAsync().ConfigureAwait(false);
+    }
 
     /// <inheritdoc/>
     public override Task<IExternalKernelProcessMessageChannel?> GetExternalMessageChannelAsync()

dotnet/src/Experimental/Process.LocalRuntime/LocalMap.cs

@@ -98,7 +98,7 @@ internal override async Task HandleMessageAsync(ProcessMessage message)
         {
             foreach (var operation in mapOperations)
             {
-                operation.ProcessContext.Dispose();
+                await operation.ProcessContext.DisposeAsync().ConfigureAwait(false);
             }
         }
     }

dotnet/src/Experimental/Process.LocalRuntime/LocalProcess.cs

@@ -16,7 +16,7 @@ namespace Microsoft.SemanticKernel;
 
 internal delegate bool ProcessEventProxy(ProcessEvent processEvent);
 
-internal sealed class LocalProcess : LocalStep, IDisposable
+internal sealed class LocalProcess : LocalStep, System.IAsyncDisposable
 {
     private readonly JoinableTaskFactory _joinableTaskFactory;
     private readonly JoinableTaskContext _joinableTaskContext;
@@ -49,8 +49,15 @@ internal LocalProcess(KernelProcess process, Kernel kernel)
         this._joinableTaskContext = new JoinableTaskContext();
         this._joinableTaskFactory = new JoinableTaskFactory(this._joinableTaskContext);
         this._logger = this._kernel.LoggerFactory?.CreateLogger(this.Name) ?? new NullLogger<LocalStep>();
+        // if parent id is null this is the root process
+        this.RootProcessId = this.ParentProcessId == null ? this.Id : null;
     }
 
+    /// <summary>
+    /// The Id of the root process.
+    /// </summary>
+    internal string? RootProcessId { get; init; }
+
     /// <summary>
     /// Starts the process with an initial event and an optional kernel.
     /// </summary>
@@ -196,6 +203,7 @@ private ValueTask InitializeProcessAsync()
                     new LocalProcess(processStep, this._kernel)
                     {
                         ParentProcessId = this.Id,
+                        RootProcessId = this.RootProcessId,
                         EventProxy = this.EventProxy,
                         ExternalMessageChannel = this.ExternalMessageChannel,
                     };
@@ -213,7 +221,7 @@ private ValueTask InitializeProcessAsync()
                 localStep =
                     new LocalProxy(proxyStep, this._kernel)
                     {
-                        ParentProcessId = this.Id,
+                        ParentProcessId = this.RootProcessId,
                         EventProxy = this.EventProxy,
                         ExternalMessageChannel = this.ExternalMessageChannel,
                     };
@@ -403,11 +411,20 @@ internal override async Task<KernelProcessStepInfo> ToKernelProcessStepInfoAsync
 
     #endregion
 
-    public void Dispose()
+    /// <inheritdoc/>
+    public override async Task DeinitializeStepAsync()
+    {
+        await this.DisposeAsync().ConfigureAwait(false);
+    }
+
+    public async ValueTask DisposeAsync()
     {
         this._externalEventChannel.Writer.Complete();
         this._joinableTaskContext.Dispose();
-        this._joinableTaskContext.Dispose();
+        foreach (var step in this._steps)
+        {
+            await step.DeinitializeStepAsync().ConfigureAwait(false);
+        }
         this._processCancelSource?.Dispose();
     }
 }

dotnet/src/Experimental/Process.LocalRuntime/LocalProxy.cs

@@ -1,9 +1,11 @@
 // Copyright (c) Microsoft. All rights reserved.
 using System.Collections.Generic;
 using System.Linq;
+using System.Reflection;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.SemanticKernel.Process;
 using Microsoft.SemanticKernel.Process.Internal;
 using Microsoft.SemanticKernel.Process.Runtime;
 
@@ -77,4 +79,28 @@ protected override async ValueTask InitializeStepAsync()
         await base.InitializeStepAsync().ConfigureAwait(false);
         this._isInitialized = true;
     }
+
+    /// <summary>
+    /// Deinitialization of the Proxy Step, calling <see cref="KernelProxyStep.DeactivateAsync(KernelProcessStepExternalContext)"/>
+    /// </summary>
+    /// <returns></returns>
+    public override async Task DeinitializeStepAsync()
+    {
+        MethodInfo? derivedMethod = this._stepInfo.InnerStepType.GetMethod(
+            nameof(KernelProxyStep.DeactivateAsync),
+            BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance,
+            binder: null,
+            types: [typeof(KernelProcessStepExternalContext)],
+            modifiers: null);
+
+        if (derivedMethod != null && this._stepInstance != null)
+        {
+            var context = new KernelProcessStepExternalContext(this.ExternalMessageChannel);
+            ValueTask deactivateTask =
+                (ValueTask?)derivedMethod.Invoke(this._stepInstance, [context]) ??
+                throw new KernelException($"The derived DeactivateAsync method failed to complete for step {this.Name}.").Log(this._logger);
+
+            await deactivateTask.ConfigureAwait(false);
+        }
+    }
 }