Merge pull request #1 from esttenorio/estenori/processes/expSerialization

esttenorio · web-flow · commit f072057de6f6 · 2025-04-02T09:21:34.000-07:00
serializing objects internally - dapr working
diff --git a/dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc/Clients/DocumentGenerationGrpcClient.cs b/dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Grpc/Clients/DocumentGenerationGrpcClient.cs
@@ -42,7 +42,7 @@ public async Task EmitExternalEventAsync(string externalTopicEvent, KernelProces
             switch (externalTopicEvent)
             {
                 case DocumentGenerationProcess.DocGenerationTopics.RequestUserReview:
-                    var requestDocument = JsonSerializer.Deserialize<DocumentInfo>(message.EventData.ToString()!);
+                    var requestDocument = message.EventData.ToObject() as DocumentInfo;
                     if (requestDocument != null)
                     {
                         await this._grpcClient.RequestUserReviewDocumentationFromProcessAsync(new()
@@ -56,7 +56,7 @@ await this._grpcClient.RequestUserReviewDocumentationFromProcessAsync(new()
                     return;
 
                 case DocumentGenerationProcess.DocGenerationTopics.PublishDocumentation:
-                    var publishedDocument = JsonSerializer.Deserialize<DocumentInfo>(message.EventData.ToString()!);
+                    var publishedDocument = message.EventData.ToObject() as DocumentInfo;
                     if (publishedDocument != null)
                     {
                         await this._grpcClient.PublishDocumentationAsync(new()
diff --git a/dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Processes/Models/DocumentInfo.cs b/dotnet/samples/Demos/ProcessWithCloudEvents/ProcessWithCloudEvents.Processes/Models/DocumentInfo.cs
@@ -10,23 +10,23 @@ namespace ProcessWithCloudEvents.Processes.Models;
 /// Since this object is used as parameter and state type by multiple steps,
 /// Its members must be public and serializable
 /// </summary>
-[DataContract]
+//[DataContract]
 public class DocumentInfo
 {
     /// <summary>
     /// Id of the document
     /// </summary>
-    [DataMember]
+    //[DataMember]
     public string Id { get; set; } = string.Empty;
     /// <summary>
     /// Title of the document
     /// </summary>
-    [DataMember]
+    //[DataMember]
     public string Title { get; set; } = string.Empty;
     /// <summary>
     /// Content of the document
     /// </summary>
-    [DataMember]
+    //[DataMember]
     public string Content { get; set; } = string.Empty;
 
     /// <summary>
@@ -35,8 +35,8 @@ public class DocumentInfo
     /// Overriding the ToString method allows injecting custom serialization logic if needed.
     /// </summary>
     /// <returns></returns>
-    public override string ToString()
-    {
-        return JsonSerializer.Serialize(this);
-    }
+    //public override string ToString()
+    //{
+    //    return JsonSerializer.Serialize(this);
+    //}
 }
diff --git a/dotnet/src/Experimental/Process.Abstractions/KernelProcessEventData.cs b/dotnet/src/Experimental/Process.Abstractions/KernelProcessEventData.cs
@@ -0,0 +1,79 @@
+﻿// Copyright (c) Microsoft. All rights reserved.
+using System;
+using System.Runtime.Serialization;
+using System.Text.Json;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// A serializable representation of an internal message used in a process runtime received by proxy steps.
+/// </summary>
+/// <remarks>
+/// Initializes a new instance of the <see cref="KernelProcessEventData"/> class.
+/// </remarks>
+[DataContract]
+public sealed record KernelProcessEventData
+{
+    /// <summary>
+    /// The assembly qualified name of the object type
+    /// </summary>
+    [DataMember]
+    public string ObjectType { get; set; } = string.Empty;
+    /// <summary>
+    /// The Json serialized object
+    /// </summary>
+    [DataMember]
+    public string Content { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Converts serialized object to original object type
+    /// </summary>
+    /// <returns></returns>
+    public object? ToObject()
+    {
+        Verify.NotNullOrWhiteSpace(this.ObjectType);
+        Type? type = Type.GetType(this.ObjectType);
+        if (type != null)
+        {
+            try
+            {
+                return JsonSerializer.Deserialize(this.Content, type);
+            }
+            catch (JsonException)
+            {
+                throw new KernelException($"Cannot deserialize object {this.Content}");
+            }
+        }
+
+        return null;
+    }
+
+    /// <summary>
+    /// Converts from original object to serialized version of the object
+    /// </summary>
+    /// <param name="obj">object to be serialized</param>
+    /// <returns>instance of <see cref="KernelProcessEventData"/></returns>
+    public static KernelProcessEventData? FromObject(object? obj)
+    {
+        if (obj == null)
+        {
+            return null;
+        }
+
+        Verify.NotNull(obj.GetType());
+        Verify.NotNull(obj.GetType().AssemblyQualifiedName);
+
+        try
+        {
+            return new KernelProcessEventData()
+            {
+                ObjectType = obj.GetType().AssemblyQualifiedName!,
+                Content = JsonSerializer.Serialize(obj)
+            };
+        }
+        catch (NotSupportedException)
+        {
+            throw new KernelException($"Cannot serialize object {obj.GetType().FullName}");
+        }
+    }
+}
diff --git a/dotnet/src/Experimental/Process.Abstractions/KernelProcessProxyMessage.cs b/dotnet/src/Experimental/Process.Abstractions/KernelProcessProxyMessage.cs
@@ -44,5 +44,5 @@ public sealed record KernelProcessProxyMessage
     /// </summary>
     [DataMember]
     [JsonPropertyName("eventData")]
-    public object? EventData { get; init; }
+    public KernelProcessEventData? EventData { get; init; }
 }
diff --git a/dotnet/src/Experimental/Process.Abstractions/KernelProcessStepExternalContext.cs b/dotnet/src/Experimental/Process.Abstractions/KernelProcessStepExternalContext.cs
@@ -33,11 +33,8 @@ public async Task EmitExternalEventAsync(KernelProcessProxyMessage processEventD
         {
             throw new KernelException($"External message channel not configured for step with topic {processEventData.ExternalTopicName}");
         }
-        // External message must be serialized to be Dapr compatible.
-        // This assumes custom serialization was provided by overriding the .ToString method in case the data is a complex object
-        var externalMessageData = processEventData with { EventData = processEventData.EventData?.ToString() };
 
-        await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, externalMessageData).ConfigureAwait(false);
+        await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, processEventData).ConfigureAwait(false);
     }
 
     /// <summary>
diff --git a/dotnet/src/Experimental/Process.IntegrationTests.Shared/ProcessCloudEventsTests.cs b/dotnet/src/Experimental/Process.IntegrationTests.Shared/ProcessCloudEventsTests.cs
@@ -73,10 +73,10 @@ public async Task LinearProcessWithCloudEventSubscribersUsingEmitToTopicAsync()
         Assert.NotNull(mockClient);
         Assert.True(mockClient.InitializationCounter > 0);
         Assert.Equal(2, mockClient.CloudEvents.Count);
-        Assert.Equal(runningProcessId, mockClient.CloudEvents[0].Data?.ProcessId);
-        Assert.Equal(runningProcessId, mockClient.CloudEvents[1].Data?.ProcessId);
-        this.AssertProxyMessage(mockClient.CloudEvents[0].Data, expectedPublishTopic: MockTopicNames.EchoExternalTopic, expectedTopicData: testInput);
-        this.AssertProxyMessage(mockClient.CloudEvents[1].Data, expectedPublishTopic: MockTopicNames.RepeatExternalTopic, expectedTopicData: $"{testInput} {testInput}");
+        Assert.Equal(runningProcessId, mockClient.CloudEvents[0].ProcessId);
+        Assert.Equal(runningProcessId, mockClient.CloudEvents[1].ProcessId);
+        this.AssertProxyMessage(mockClient.CloudEvents[0], expectedPublishTopic: MockTopicNames.EchoExternalTopic, expectedTopicData: testInput);
+        this.AssertProxyMessage(mockClient.CloudEvents[1], expectedPublishTopic: MockTopicNames.RepeatExternalTopic, expectedTopicData: $"{testInput} {testInput}");
     }
 
     /// <summary>
@@ -111,15 +111,15 @@ public async Task ProcessWithSubprocessWithProxyEmittingDifferentTopicsAsync()
         Assert.NotNull(mockClient);
         Assert.True(mockClient.InitializationCounter > 0);
         Assert.Equal(2, mockClient.CloudEvents.Count);
-        if (mockClient.CloudEvents[0].TopicName == this._topic1)
+        if (mockClient.CloudEvents[0].ExternalTopicName == this._topic1)
         {
-            this.AssertProxyMessage(mockClient.CloudEvents[0].Data, expectedPublishTopic: this._topic1, expectedTopicData: testInput);
-            this.AssertProxyMessage(mockClient.CloudEvents[1].Data, expectedPublishTopic: this._topic2, expectedTopicData: testInput);
+            this.AssertProxyMessage(mockClient.CloudEvents[0], expectedPublishTopic: this._topic1, expectedTopicData: testInput);
+            this.AssertProxyMessage(mockClient.CloudEvents[1], expectedPublishTopic: this._topic2, expectedTopicData: testInput);
         }
         else
         {
-            this.AssertProxyMessage(mockClient.CloudEvents[0].Data, expectedPublishTopic: this._topic2, expectedTopicData: testInput);
-            this.AssertProxyMessage(mockClient.CloudEvents[1].Data, expectedPublishTopic: this._topic1, expectedTopicData: testInput);
+            this.AssertProxyMessage(mockClient.CloudEvents[0], expectedPublishTopic: this._topic2, expectedTopicData: testInput);
+            this.AssertProxyMessage(mockClient.CloudEvents[1], expectedPublishTopic: this._topic1, expectedTopicData: testInput);
         }
     }
 
@@ -155,8 +155,8 @@ public async Task ProcessWithSubprocessWithProxyEmittingSameTopicsAsync()
         Assert.NotNull(mockClient);
         Assert.True(mockClient.InitializationCounter > 0);
         Assert.Equal(2, mockClient.CloudEvents.Count);
-        this.AssertProxyMessage(mockClient.CloudEvents[0].Data, expectedPublishTopic: this._topic1, expectedTopicData: testInput);
-        this.AssertProxyMessage(mockClient.CloudEvents[1].Data, expectedPublishTopic: this._topic1, expectedTopicData: testInput);
+        this.AssertProxyMessage(mockClient.CloudEvents[0], expectedPublishTopic: this._topic1, expectedTopicData: testInput);
+        this.AssertProxyMessage(mockClient.CloudEvents[1], expectedPublishTopic: this._topic1, expectedTopicData: testInput);
     }
 
     /// <summary>
@@ -225,17 +225,11 @@ private void AssertProxyMessage(KernelProcessProxyMessage? proxyMessage, string
         Assert.NotNull(proxyMessage);
         Assert.IsType<KernelProcessProxyMessage>(proxyMessage);
         Assert.Equal(expectedPublishTopic, proxyMessage.ExternalTopicName);
-        if (proxyMessage.EventData is JsonElement jsonEventData)
-        {
-            // needed for Dapr Testing setup since it serializes everything with json
-            Assert.Equal(JsonValueKind.String, jsonEventData.ValueKind);
-            Assert.Equal(expectedTopicData, jsonEventData.ToString());
-        }
-        else
-        {
-            Assert.IsType<string>(proxyMessage.EventData);
-            Assert.Equal(expectedTopicData, proxyMessage.EventData);
-        }
+
+        Assert.IsType<KernelProcessEventData>(proxyMessage.EventData);
+        var outputEventData = proxyMessage.EventData.ToObject();
+        Assert.IsType<string>(outputEventData);
+        Assert.Equal(expectedTopicData, outputEventData);
     }
     #endregion
 }
diff --git a/dotnet/src/Experimental/Process.LocalRuntime/LocalStep.cs b/dotnet/src/Experimental/Process.LocalRuntime/LocalStep.cs
@@ -154,7 +154,14 @@ internal virtual void AssignStepFunctionParameterValues(ProcessMessage message)
                 functionParameters = this._inputs[message.FunctionName];
             }
 
-            functionParameters![kvp.Key] = kvp.Value;
+            if (kvp.Value is KernelProcessEventData proxyData)
+            {
+                functionParameters![kvp.Key] = proxyData.ToObject();
+            }
+            else
+            {
+                functionParameters![kvp.Key] = kvp.Value;
+            }
         }
     }
 
@@ -212,24 +219,22 @@ internal virtual async Task HandleMessageAsync(ProcessMessage message)
         {
             FunctionResult invokeResult = await this.InvokeFunction(function, this._kernel, arguments).ConfigureAwait(false);
             this.EmitEvent(
-                new ProcessEvent
-                {
-                    Namespace = this._eventNamespace,
-                    SourceId = $"{targetFunction}.OnResult",
-                    Data = invokeResult.GetValue<object>()
-                });
+                ProcessEvent.Create(
+                    invokeResult.GetValue<object>(),
+                    this._eventNamespace,
+                    sourceId: $"{targetFunction}.OnResult",
+                    eventVisibility: KernelProcessEventVisibility.Public));
         }
         catch (Exception ex)
         {
             this._logger.LogError(ex, "Error in Step {StepName}: {ErrorMessage}", this.Name, ex.Message);
             this.EmitEvent(
-                new ProcessEvent
-                {
-                    Namespace = this._eventNamespace,
-                    SourceId = $"{targetFunction}.OnError",
-                    Data = KernelProcessError.FromException(ex),
-                    IsError = true
-                });
+                ProcessEvent.Create(
+                    KernelProcessError.FromException(ex),
+                    this._eventNamespace,
+                    sourceId: $"{targetFunction}.OnError",
+                    eventVisibility: KernelProcessEventVisibility.Public,
+                    isError: true));
         }
         finally
         {
diff --git a/dotnet/src/Experimental/Process.Runtime.Dapr/Actors/StepActor.cs b/dotnet/src/Experimental/Process.Runtime.Dapr/Actors/StepActor.cs
@@ -234,7 +234,14 @@ internal virtual void AssignStepFunctionParameterValues(ProcessMessage message)
                 functionParameters = this._inputs[message.FunctionName];
             }
 
-            functionParameters![kvp.Key] = kvp.Value;
+            if (kvp.Value is KernelProcessEventData proxyData)
+            {
+                functionParameters![kvp.Key] = proxyData.ToObject();
+            }
+            else
+            {
+                functionParameters![kvp.Key] = kvp.Value;
+            }
         }
     }
 
@@ -301,24 +308,22 @@ internal virtual async Task HandleMessageAsync(ProcessMessage message)
             await this.StateManager.SaveStateAsync().ConfigureAwait(false);
 
             await this.EmitEventAsync(
-                new ProcessEvent
-                {
-                    Namespace = this._eventNamespace!,
-                    SourceId = $"{targetFunction}.OnResult",
-                    Data = invokeResult.GetValue<object>()
-                }).ConfigureAwait(false);
+                ProcessEvent.Create(
+                    invokeResult.GetValue<object>(),
+                    this._eventNamespace!,
+                    sourceId: $"{targetFunction}.OnResult",
+                    eventVisibility: KernelProcessEventVisibility.Public)).ConfigureAwait(false);
         }
         catch (Exception ex)
         {
             this._logger?.LogError(ex, "Error in Step {StepName}: {ErrorMessage}", this.Name, ex.Message);
             await this.EmitEventAsync(
-                new ProcessEvent
-                {
-                    Namespace = this._eventNamespace!,
-                    SourceId = $"{targetFunction}.OnError",
-                    Data = KernelProcessError.FromException(ex),
-                    IsError = true
-                }).ConfigureAwait(false);
+                ProcessEvent.Create(
+                    KernelProcessError.FromException(ex),
+                    this._eventNamespace!,
+                    sourceId: $"{targetFunction}.OnError",
+                    eventVisibility: KernelProcessEventVisibility.Public,
+                    isError: true)).ConfigureAwait(false);
         }
         finally
         {
diff --git a/dotnet/src/Experimental/Process.UnitTests/Runtime.Local/LocalProxyTests.cs b/dotnet/src/Experimental/Process.UnitTests/Runtime.Local/LocalProxyTests.cs
diff --git a/dotnet/src/InternalUtilities/process/Runtime/KernelProcessProxyMessageFactory.cs b/dotnet/src/InternalUtilities/process/Runtime/KernelProcessProxyMessageFactory.cs
diff --git a/dotnet/src/InternalUtilities/process/Runtime/ProcessEvent.cs b/dotnet/src/InternalUtilities/process/Runtime/ProcessEvent.cs
diff --git a/dotnet/src/InternalUtilities/process/TestsShared/CloudEvents/MockCloudEventClient.cs b/dotnet/src/InternalUtilities/process/TestsShared/CloudEvents/MockCloudEventClient.cs

Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ public async Task EmitExternalEventAsync(string externalTopicEvent, KernelProces`
`42`	`42`	`switch (externalTopicEvent)`
`43`	`43`	`{`
`44`	`44`	`case DocumentGenerationProcess.DocGenerationTopics.RequestUserReview:`
`45`		`- var requestDocument = JsonSerializer.Deserialize<DocumentInfo>(message.EventData.ToString()!);`
	`45`	`+ var requestDocument = message.EventData.ToObject() as DocumentInfo;`
`46`	`46`	`if (requestDocument != null)`
`47`	`47`	`{`
`48`	`48`	`await this._grpcClient.RequestUserReviewDocumentationFromProcessAsync(new()`
`@@ -56,7 +56,7 @@ await this._grpcClient.RequestUserReviewDocumentationFromProcessAsync(new()`
`56`	`56`	`return;`
`57`	`57`
`58`	`58`	`case DocumentGenerationProcess.DocGenerationTopics.PublishDocumentation:`
`59`		`- var publishedDocument = JsonSerializer.Deserialize<DocumentInfo>(message.EventData.ToString()!);`
	`59`	`+ var publishedDocument = message.EventData.ToObject() as DocumentInfo;`
`60`	`60`	`if (publishedDocument != null)`
`61`	`61`	`{`
`62`	`62`	`await this._grpcClient.PublishDocumentationAsync(new()`
Original file line number	Diff line number	Diff line change
`@@ -44,5 +44,5 @@ public sealed record KernelProcessProxyMessage`
`44`	`44`	`/// </summary>`
`45`	`45`	`[DataMember]`
`46`	`46`	`[JsonPropertyName("eventData")]`
`47`		`- public object? EventData { get; init; }`
	`47`	`+ public KernelProcessEventData? EventData { get; init; }`
`48`	`48`	`}`
Original file line number	Diff line number	Diff line change
`@@ -33,11 +33,8 @@ public async Task EmitExternalEventAsync(KernelProcessProxyMessage processEventD`
`33`	`33`	`{`
`34`	`34`	`throw new KernelException($"External message channel not configured for step with topic {processEventData.ExternalTopicName}");`
`35`	`35`	`}`
`36`		`- // External message must be serialized to be Dapr compatible.`
`37`		`- // This assumes custom serialization was provided by overriding the .ToString method in case the data is a complex object`
`38`		`- var externalMessageData = processEventData with { EventData = processEventData.EventData?.ToString() };`
`39`	`36`
`40`		`- await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, externalMessageData).ConfigureAwait(false);`
	`37`	`+ await this._externalMessageChannel.EmitExternalEventAsync(processEventData.ExternalTopicName, processEventData).ConfigureAwait(false);`
`41`	`38`	`}`
`42`	`39`
`43`	`40`	`/// <summary>`