.Net: Processes - Tests for OnFunctionError events. (#9703)

### Description

This PR adds one new integration test for processes where a step throws
and exception and we expect the framework to emit an OnFunctionError
event.

There are other minor changes to improve the de/serialization of events
within the shared test infrastructure.

### Contribution Checklist

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

- [x] The code builds clean without any errors or warnings
- [x] 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
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄

---------

Co-authored-by: Chris <[email protected]>

dotnet/src/Experimental/Process.IntegrationTestHost.Dapr/Contracts/ProcessStartRequest.cs

@@ -17,5 +17,5 @@ public record ProcessStartRequest
     /// <summary>
     /// The initial event to send to the process.
     /// </summary>
-    public required KernelProcessEvent InitialEvent { get; set; }
+    public required string InitialEvent { get; set; }
 }

dotnet/src/Experimental/Process.IntegrationTestHost.Dapr/Controllers/ProcessTestController.cs

@@ -4,6 +4,7 @@
 using Dapr.Actors.Client;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.SemanticKernel;
+using Microsoft.SemanticKernel.Process.Serialization;
 
 namespace SemanticKernel.Process.IntegrationTests.Controllers;
 
@@ -41,14 +42,10 @@ public async Task<IActionResult> StartProcessAsync(string processId, [FromBody]
             return this.BadRequest("Process already started");
         }
 
-        if (request.InitialEvent?.Data is JsonElement jsonElement)
-        {
-            object? data = jsonElement.Deserialize<string>();
-            request.InitialEvent = request.InitialEvent with { Data = data };
-        }
+        var initialEvent = KernelProcessEventSerializer.ToKernelProcessEvent(request.InitialEvent);
 
         var kernelProcess = request.Process.ToKernelProcess();
-        var context = await kernelProcess.StartAsync(request.InitialEvent!);
+        var context = await kernelProcess.StartAsync(initialEvent);
         s_processes.Add(processId, context);
 
         return this.Ok();

dotnet/src/Experimental/Process.IntegrationTestRunner.Dapr/DaprTestProcessContext.cs

@@ -4,6 +4,7 @@
 using System.Text.Json;
 using Microsoft.SemanticKernel;
 using Microsoft.SemanticKernel.Process;
+using Microsoft.SemanticKernel.Process.Serialization;
 
 namespace SemanticKernel.Process.IntegrationTests;
 internal sealed class DaprTestProcessContext : KernelProcessContext
@@ -39,7 +40,7 @@ internal DaprTestProcessContext(KernelProcess process, HttpClient httpClient)
     internal async Task StartWithEventAsync(KernelProcessEvent initialEvent)
     {
         var daprProcess = DaprProcessInfo.FromKernelProcess(this._process);
-        var request = new ProcessStartRequest { Process = daprProcess, InitialEvent = initialEvent };
+        var request = new ProcessStartRequest { Process = daprProcess, InitialEvent = initialEvent.ToJson() };
 
         var response = await this._httpClient.PostAsJsonAsync($"http://localhost:5200/processes/{this._processId}", request, options: this._serializerOptions).ConfigureAwait(false);
         if (!response.IsSuccessStatusCode)

dotnet/src/Experimental/Process.IntegrationTests.Resources/ProcessCycleTestResources.cs

@@ -202,6 +202,60 @@ await context.EmitEventAsync(new()
     }
 }
 
+/// <summary>
+/// A step that conditionally throws an exception.
+/// </summary>
+public sealed class ErrorStep : KernelProcessStep<StepState>
+{
+    private StepState? _state;
+
+    public override ValueTask ActivateAsync(KernelProcessStepState<StepState> state)
+    {
+        this._state = state.State;
+        return default;
+    }
+
+    [KernelFunction]
+    public async Task ErrorWhenTrueAsync(KernelProcessStepContext context, bool shouldError)
+    {
+        this._state!.InvocationCount++;
+
+        if (shouldError)
+        {
+            throw new InvalidOperationException("This is an error");
+        }
+
+        await context.EmitEventAsync(new()
+        {
+            Id = ProcessTestsEvents.ErrorStepSuccess,
+            Data = null,
+            Visibility = KernelProcessEventVisibility.Internal
+        });
+    }
+}
+
+/// <summary>
+/// A step that reports an error sent to it by logging it to the console.
+/// </summary>
+public sealed class ReportStep : KernelProcessStep<StepState>
+{
+    private StepState? _state;
+
+    public override ValueTask ActivateAsync(KernelProcessStepState<StepState> state)
+    {
+        this._state = state.State;
+        return default;
+    }
+
+    [KernelFunction]
+    public Task ReportError(KernelProcessStepContext context, object error)
+    {
+        this._state!.InvocationCount++;
+        Console.WriteLine(error.ToString());
+        return Task.CompletedTask;
+    }
+}
+
 /// <summary>
 /// The state object for the repeat and fanIn step.
 /// </summary>
@@ -210,6 +264,9 @@ public sealed record StepState
 {
     [DataMember]
     public string? LastMessage { get; set; }
+
+    [DataMember]
+    public int InvocationCount { get; set; }
 }
 
 /// <summary>
@@ -222,6 +279,7 @@ public static class ProcessTestsEvents
     public const string StartInnerProcess = "StartInnerProcess";
     public const string OutputReadyPublic = "OutputReadyPublic";
     public const string OutputReadyInternal = "OutputReadyInternal";
+    public const string ErrorStepSuccess = "ErrorStepSuccess";
 }
 
 #pragma warning restore CS1591 // Missing XML comment for publicly visible type or member

dotnet/src/Experimental/Process.IntegrationTests.Shared/ProcessTests.cs

@@ -217,6 +217,32 @@ public async Task FanInProcessAsync()
         Assert.Equal($"{testInput}-{testInput} {testInput}", outputStep.State.LastMessage);
     }
 
+    /// <summary>
+    /// Test with a process that has an error step that emits an error event
+    /// </summary>
+    /// <returns></returns>
+    [Fact]
+    public async Task ProcessWithErrorEmitsErrorEventAsync()
+    {
+        // Arrange
+        Kernel kernel = this._kernelBuilder.Build();
+        var process = this.CreateProcessWithError("ProcessWithError").Build();
+
+        // Act
+        bool shouldError = true;
+        var processHandle = await this._fixture.StartProcessAsync(process, kernel, new() { Id = ProcessTestsEvents.StartProcess, Data = shouldError });
+        var processInfo = await processHandle.GetStateAsync();
+
+        // Assert
+        var reportStep = processInfo.Steps.Where(s => s.State.Name == nameof(ReportStep)).FirstOrDefault()?.State as KernelProcessStepState<StepState>;
+        Assert.NotNull(reportStep?.State);
+        Assert.Equal(1, reportStep.State.InvocationCount);
+
+        var repeatStep = processInfo.Steps.Where(s => s.State.Name == nameof(RepeatStep)).FirstOrDefault()?.State as KernelProcessStepState<StepState>;
+        Assert.NotNull(repeatStep?.State);
+        Assert.Null(repeatStep.State.LastMessage);
+    }
+
     /// <summary>
     /// Test with a single step that then connects to a nested fan in process with 2 input steps
     /// </summary>
@@ -280,4 +306,18 @@ private ProcessBuilder CreateFanInProcess(string name)
 
         return processBuilder;
     }
+
+    private ProcessBuilder CreateProcessWithError(string name)
+    {
+        var processBuilder = new ProcessBuilder(name);
+        var errorStep = processBuilder.AddStepFromType<ErrorStep>("ErrorStep");
+        var repeatStep = processBuilder.AddStepFromType<RepeatStep>("RepeatStep");
+        var reportStep = processBuilder.AddStepFromType<ReportStep>("ReportStep");
+
+        processBuilder.OnInputEvent(ProcessTestsEvents.StartProcess).SendEventTo(new ProcessFunctionTargetBuilder(errorStep));
+        errorStep.OnEvent(ProcessTestsEvents.ErrorStepSuccess).SendEventTo(new ProcessFunctionTargetBuilder(repeatStep, parameterName: "message"));
+        errorStep.OnFunctionError("ErrorWhenTrue").SendEventTo(new ProcessFunctionTargetBuilder(reportStep));
+
+        return processBuilder;
+    }
 }

dotnet/src/Experimental/Process.Runtime.Dapr/Process.Runtime.Dapr.csproj

@@ -27,6 +27,8 @@
 
   <ItemGroup>
     <InternalsVisibleTo Include="SemanticKernel.Process.Runtime.Dapr.UnitTests" />
+    <InternalsVisibleTo Include="SemanticKernel.Process.IntegrationTestHost.Dapr" />
+    <InternalsVisibleTo Include="SemanticKernel.Process.IntegrationTestRunner.Dapr" />
     <InternalsVisibleTo Include="DynamicProxyGenAssembly2" />
   </ItemGroup>