.Net: Process Cloud Events - Publish Events Abstractions

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

@@ -28,5 +28,5 @@ public interface IExternalKernelProcessMessageChannel
     /// <param name="externalTopicEvent">name of the topic to be used externally as the event name</param>
     /// <param name="eventData">data to be transmitted externally</param>
     /// <returns></returns>
-    abstract Task EmitExternalEventAsync(string externalTopicEvent, object? eventData);
+    public abstract Task EmitExternalEventAsync(string externalTopicEvent, KernelProcessProxyMessage eventData);
 }

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

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Collections.Generic;
+using Microsoft.SemanticKernel.Process;
+using Microsoft.SemanticKernel.Process.Models;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// A serializable representation of a ProcessProxy.
+/// </summary>
+public sealed record KernelProcessProxy : KernelProcessStepInfo
+{
+    /// <summary>
+    /// Proxy metadata used for linking specific SK events to external events and viceversa
+    /// </summary>
+    public KernelProcessProxyStateMetadata? ProxyMetadata { get; init; }
+
+    /// <summary>
+    /// Creates a new instance of the <see cref="KernelProcess"/> class.
+    /// </summary>
+    /// <param name="state">The process state.</param>
+    /// <param name="edges">The edges for the map.</param>
+    public KernelProcessProxy(KernelProcessStepState state, Dictionary<string, List<KernelProcessEdge>> edges)
+        : base(typeof(KernelProxyStep), state, edges)
+    {
+        Verify.NotNullOrWhiteSpace(state.Name, $"{nameof(state)}.{nameof(KernelProcessStepState.Name)}");
+        Verify.NotNullOrWhiteSpace(state.Id, $"{nameof(state)}.{nameof(KernelProcessStepState.Id)}");
+    }
+}

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

@@ -0,0 +1,48 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Runtime.Serialization;
+using System.Text.Json.Serialization;
+
+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="KernelProcessProxyMessage"/> class.
+/// </remarks>
+[DataContract]
+public record KernelProcessProxyMessage
+{
+    /// <summary>
+    /// Id of the SK process that emits the external event
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("processId")]
+    public string? ProcessId { get; init; }
+
+    /// <summary>
+    /// Name of the SK process that triggers sending the event externally
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("triggerEventId")]
+    public string? TriggerEventId { get; init; }
+
+    /// <summary>
+    /// Topic name used for publishing process event data externally
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("externalTopicName")]
+    public string? ExternalTopicName { get; init; }
+    /// <summary>
+    /// Event name used for publishing process event as another process event with a different event name
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("proxyEventName")]
+    public string? ProxyEventName { get; init; }
+    /// <summary>
+    /// Data to be emitted
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("eventData")]
+    public object? EventData { get; init; }
+}

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

@@ -60,11 +60,11 @@ public ValueTask EmitEventAsync(
     /// Emit an external event to through a <see cref="IExternalKernelProcessMessageChannel"/>
     /// component if connected from within the SK process
     /// </summary>
-    /// <param name="topicName"></param>
-    /// <param name="processEventData"></param>
+    /// <param name="topicName">name of the topic used for external messages</param>
+    /// <param name="processEventData">data containing event details</param>
     /// <returns></returns>
     /// <exception cref="KernelException"></exception>
-    public async Task EmitExternalEventAsync(string topicName, object? processEventData = null)
+    public async Task EmitExternalEventAsync(string topicName, KernelProcessProxyMessage processEventData)
     {
         if (this._externalMessageChannel == null)
         {

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

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Threading.Tasks;
+
+namespace Microsoft.SemanticKernel.Process;
+
+/// <summary>
+/// Internal SK KernelProcessStep preconfigured to be used when emitting SK events outside of the SK Process Framework or inside with a different event name
+/// </summary>
+public sealed class KernelProxyStep : KernelProcessStep
+{
+    /// <summary>
+    /// SK Function names in this SK Step as entry points
+    /// </summary>
+    public static class Functions
+    {
+        /// <summary>
+        /// Function name used to emit events externally
+        /// </summary>
+        public const string EmitExternalEvent = nameof(EmitExternalEvent);
+    }
+
+    /// <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)
+    {
+        Verify.NotNull(proxyEvent.ExternalTopicName, nameof(proxyEvent.ExternalTopicName));
+        return context.EmitExternalEventAsync(proxyEvent.ExternalTopicName, proxyEvent);
+    }
+}

dotnet/src/Experimental/Process.Abstractions/Models/KernelProcessProxyEventMetadata.cs

@@ -0,0 +1,25 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Runtime.Serialization;
+using System.Text.Json.Serialization;
+
+namespace Microsoft.SemanticKernel.Process.Models;
+
+/// <summary>
+/// Process state used for State Persistence serialization
+/// </summary>
+public sealed record class KernelProcessProxyEventMetadata
+{
+    /// <summary>
+    /// Name of the topic to be emitted externally
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("topicName")]
+    public string TopicName { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Internal id used to identify the SK event
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("eventId")]
+    public string EventId { get; set; } = string.Empty;
+}

dotnet/src/Experimental/Process.Abstractions/Models/KernelProcessProxyStateMetadata.cs

@@ -0,0 +1,26 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Collections.Generic;
+using System.Runtime.Serialization;
+using System.Text.Json.Serialization;
+
+namespace Microsoft.SemanticKernel.Process.Models;
+
+/// <summary>
+/// Process state used for State Persistence serialization
+/// </summary>
+public sealed record class KernelProcessProxyStateMetadata : KernelProcessStepStateMetadata
+{
+    /// <summary>
+    /// List of publish topics that can be used by the SK process
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("publishTopics")]
+    public List<string> PublishTopics { get; set; } = [];
+
+    /// <summary>
+    /// Map that stores which process events trigger external topic to be published and internal metadata information
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("eventMetadata")]
+    public Dictionary<string, KernelProcessProxyEventMetadata> EventMetadata { get; set; } = [];
+}

dotnet/src/Experimental/Process.Abstractions/Models/KernelProcessStepStateMetadata.cs

@@ -12,6 +12,7 @@ namespace Microsoft.SemanticKernel.Process.Models;
 [JsonPolymorphic(TypeDiscriminatorPropertyName = "$type", UnknownDerivedTypeHandling = JsonUnknownDerivedTypeHandling.FallBackToNearestAncestor)]
 [JsonDerivedType(typeof(KernelProcessStepStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Step))]
 [JsonDerivedType(typeof(KernelProcessMapStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Map))]
+[JsonDerivedType(typeof(KernelProcessProxyStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Proxy))]
 [JsonDerivedType(typeof(KernelProcessStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Process))]
 public record class KernelProcessStepStateMetadata
 {

dotnet/src/Experimental/Process.Core/Internal/ProcessEventData.cs

@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+namespace Microsoft.SemanticKernel.Process.Internal;
+internal class ProcessEventData
+{
+    /// <summary>
+    /// SK Process Event Id, id assigned during runtime
+    /// </summary>
+    public string EventId { get; init; } = string.Empty;
+
+    /// <summary>
+    /// SK Process Event Name, human readable, defined when creating the process builder
+    /// </summary>
+    public string EventName { get; init; } = string.Empty;
+}

dotnet/src/Experimental/Process.Core/ProcessBuilder.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using Microsoft.SemanticKernel.Process;
 using Microsoft.SemanticKernel.Process.Internal;
 using Microsoft.SemanticKernel.Process.Models;
 
@@ -242,6 +243,22 @@ public ProcessMapBuilder AddMapStepFromProcess(ProcessBuilder process, IReadOnly
         return this.AddStep(mapBuilder, aliases);
     }
 
+    /// <summary>
+    /// Adds proxy step to the process that allows emitting events externally. For making use of it, there should be an implementation
+    /// of <see cref="IExternalKernelProcessMessageChannel"/> passed.
+    /// For now, the current implementation only allows for 1 implementation of <see cref="IExternalKernelProcessMessageChannel"/> at the time.
+    /// </summary>
+    /// <param name="externalTopics">topic names to be used externally</param>
+    /// <param name="name">name of the proxy step</param>
+    /// <param name="aliases">Aliases that have been used by previous versions of the step, used for supporting backward compatibility when reading old version Process States</param>
+    /// <returns>An instance of <see cref="ProcessProxyBuilder"/></returns>
+    public ProcessProxyBuilder AddProxyStep(IReadOnlyList<string> externalTopics, string? name = null, IReadOnlyList<string>? aliases = null)
+    {
+        ProcessProxyBuilder proxyBuilder = new(externalTopics, name ?? nameof(KernelProxyStep));
+
+        return this.AddStep(proxyBuilder, aliases);
+    }
+
     /// <summary>
     /// Provides an instance of <see cref="ProcessStepEdgeBuilder"/> for defining an edge to a
     /// step inside the process for a given external event.

dotnet/src/Experimental/Process.Core/ProcessEdgeBuilder.cs

@@ -46,7 +46,7 @@ public ProcessEdgeBuilder SendEventTo(ProcessFunctionTargetBuilder target)
         }
 
         this.Target = target;
-        ProcessStepEdgeBuilder edgeBuilder = new(this.Source, this.EventId) { Target = this.Target };
+        ProcessStepEdgeBuilder edgeBuilder = new(this.Source, this.EventId, this.EventId) { Target = this.Target };
         this.Source.LinkTo(this.EventId, edgeBuilder);
 
         return new ProcessEdgeBuilder(this.Source, this.EventId);

dotnet/src/Experimental/Process.Core/ProcessProxyBuilder.cs

@@ -0,0 +1,93 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.SemanticKernel.Process;
+using Microsoft.SemanticKernel.Process.Internal;
+using Microsoft.SemanticKernel.Process.Models;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// Provides functionality to allow emitting external messages from within the SK
+/// process.
+/// </summary>
+public sealed class ProcessProxyBuilder : ProcessStepBuilder<KernelProxyStep>
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ProcessProxyBuilder"/> class.
+    /// </summary>
+    internal ProcessProxyBuilder(IReadOnlyList<string> externalTopics, string name)
+        : base(name)
+    {
+        if (externalTopics.Count == 0)
+        {
+            throw new ArgumentException("No topic names registered");
+        }
+
+        this.ExternalTopicUsage = externalTopics.ToDictionary(topic => topic, topic => false);
+        if (this.ExternalTopicUsage.Count < externalTopics.Count)
+        {
+            throw new ArgumentException("Topic names registered must be different");
+        }
+    }
+
+    /// <summary>
+    /// Version of the map-step, used when saving the state of the step.
+    /// </summary>
+    public string Version { get; init; } = "v1";
+
+    internal Dictionary<string, bool> ExternalTopicUsage { get; }
+
+    // For supporting multiple step edges getting linked to the same external topic, current implementation needs to be updated
+    // to instead have a list of potential edges in case event names in different steps have same name
+    internal Dictionary<string, KernelProcessProxyEventMetadata> EventMetadata { get; } = [];
+
+    internal ProcessFunctionTargetBuilder GetExternalFunctionTargetBuilder()
+    {
+        return new ProcessFunctionTargetBuilder(this, functionName: KernelProxyStep.Functions.EmitExternalEvent, parameterName: "proxyEvent");
+    }
+
+    internal void LinkTopicToStepEdgeInfo(string topicName, ProcessStepBuilder sourceStep, ProcessEventData eventData)
+    {
+        if (!this.ExternalTopicUsage.TryGetValue(topicName, out bool usedTopic))
+        {
+            throw new InvalidOperationException($"Topic name {topicName} is not registered as proxy publish event, register first before using");
+        }
+
+        if (usedTopic)
+        {
+            throw new InvalidOperationException($"Topic name {topicName} is is already linked to another step edge");
+        }
+
+        this.EventMetadata[eventData.EventName] = new() { EventId = eventData.EventId, TopicName = topicName };
+        this.ExternalTopicUsage[topicName] = true;
+    }
+
+    /// <inheritdoc/>
+    internal override KernelProcessStepInfo BuildStep(KernelProcessStepStateMetadata? stateMetadata = null)
+    {
+        if (this.ExternalTopicUsage.All(topic => !topic.Value))
+        {
+            throw new InvalidOperationException("Proxy step does not have linked steps to it, link step edges to proxy or remove proxy step");
+        }
+
+        KernelProcessProxyStateMetadata proxyMetadata = new()
+        {
+            Name = this.Name,
+            Id = this.Id,
+            EventMetadata = this.EventMetadata,
+            PublishTopics = this.ExternalTopicUsage.ToList().Select(topic => topic.Key).ToList(),
+        };
+
+        // Build the edges first
+        var builtEdges = this.Edges.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.Select(e => e.Build()).ToList());
+
+        KernelProcessStepState state = new(this.Name, this.Version, this.Id);
+
+        return new KernelProcessProxy(state, builtEdges)
+        {
+            ProxyMetadata = proxyMetadata
+        };
+    }
+}

dotnet/src/Experimental/Process.Core/ProcessStepBuilder.cs

@@ -41,7 +41,7 @@ public ProcessStepEdgeBuilder OnEvent(string eventId)
     {
         // scope the event to this instance of this step
         var scopedEventId = this.GetScopedEventId(eventId);
-        return new ProcessStepEdgeBuilder(this, scopedEventId);
+        return new ProcessStepEdgeBuilder(this, scopedEventId, eventId);
     }
 
     /// <summary>
@@ -229,7 +229,7 @@ protected ProcessStepBuilder(string name)
 /// <summary>
 /// Provides functionality for incrementally defining a process step.
 /// </summary>
-public sealed class ProcessStepBuilder<TStep> : ProcessStepBuilder where TStep : KernelProcessStep
+public class ProcessStepBuilder<TStep> : ProcessStepBuilder where TStep : KernelProcessStep
 {
     /// <summary>
     /// The initial state of the step. This may be null if the step does not have any state.