From 931204f6dd74341938fa16ec5641a483a7ced5fe Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Thu, 6 Nov 2025 15:01:04 -0500 Subject: [PATCH] WIP: Proposal for StreamingRunWatcher API --- .../EventBinding.cs | 31 +++ .../EventChain.cs | 103 +++++++++ .../IWorkflowEventChain.cs | 11 + .../StreamingRunWatcher.cs | 210 ++++++++++++++++++ 4 files changed, 355 insertions(+) create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/EventBinding.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/EventChain.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowEventChain.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRunWatcher.cs diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/EventBinding.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/EventBinding.cs new file mode 100644 index 0000000000..59839acae7 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/EventBinding.cs @@ -0,0 +1,31 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows; + +internal class EventBinding where TEvent : WorkflowEvent +{ + private readonly EventChain _eventChain; + + internal EventBinding(EventChain eventChain) + { + this._eventChain = eventChain; + } + + internal ValueTask RaiseAsync(TEvent evt, CancellationToken cancelationToken) + => this._eventChain.RaiseAsync(evt, cancelationToken); + + public event WorkflowEventHandler? Event + { + add => this._eventChain.AttachHandler(value); + remove => this._eventChain.DetachHandler(value); + } + + public event WorkflowEventHandlerAsync? AsyncEvent + { + add => this._eventChain.AttachHandler(value); + remove => this._eventChain.DetachHandler(value); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/EventChain.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/EventChain.cs new file mode 100644 index 0000000000..fc8b65889d --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/EventChain.cs @@ -0,0 +1,103 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows; + +internal class EventChain : IWorkflowEventChain, IDisposable where TEvent : WorkflowEvent +{ + private readonly SemaphoreSlim _semaphore = new(1, 1); + private readonly List>> _handlers = new(); + private readonly Dictionary>> _handlerMap = new(); + + ValueTask IWorkflowEventChain.RaiseAsync(WorkflowEvent evt, CancellationToken cancellationToken) + { + if (evt is TEvent typedEvent) + { + return this.RaiseAsync(typedEvent, cancellationToken); + } + + throw new InvalidOperationException($"Sending event of type {evt.GetType().FullName} to event chain for " + + $"type {typeof(TEvent).FullName}."); + } + + public async ValueTask RaiseAsync(TEvent evt, CancellationToken cancellationToken) + { + await this._semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + foreach (Func> handler in this._handlers) + { + bool handled = await handler(evt).ConfigureAwait(false); + if (handled) + { + return true; + } + } + + return false; + } + finally + { + this._semaphore.Release(); + } + } + + private bool AttachHandler(Func> handlerAsync, object key) + { + this._semaphore.Wait(); + + bool shouldAdd = !this._handlerMap.ContainsKey(key); + if (shouldAdd) + { + this._handlerMap.Add(key, handlerAsync); + this._handlers.Add(handlerAsync); + } + + this._semaphore.Release(); + return shouldAdd; + } + + private bool DetachHandler(object key) + { + this._semaphore.Wait(); + + bool removed = false; + if (this._handlerMap.TryGetValue(key, out Func>? handler)) + { + this._handlers.Remove(handler); + removed = true; + } + + this._semaphore.Release(); + return removed; + } + + public bool AttachHandler(WorkflowEventHandlerAsync? handlerAsync) + => handlerAsync is not null + ? this.AttachHandler(handlerAsync.Invoke, handlerAsync) + : false; + + public bool AttachHandler(WorkflowEventHandler? handler) + => handler is not null + ? this.AttachHandler(evt => new(handler.Invoke(evt)), handler) + : false; + + public bool DetachHandler(WorkflowEventHandlerAsync? handlerAsync) + => handlerAsync is not null + ? this.DetachHandler((object)handlerAsync) + : false; + + public bool DetachHandler(WorkflowEventHandler? handler) + => handler is not null + ? this.DetachHandler((object)handler) + : false; + + public void Dispose() + { + this._semaphore.Dispose(); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowEventChain.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowEventChain.cs new file mode 100644 index 0000000000..a67262f7cc --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowEventChain.cs @@ -0,0 +1,11 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows; + +internal interface IWorkflowEventChain +{ + ValueTask RaiseAsync(WorkflowEvent evt, CancellationToken cancellationToken); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRunWatcher.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRunWatcher.cs new file mode 100644 index 0000000000..4e6d8f4d63 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/StreamingRunWatcher.cs @@ -0,0 +1,210 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows; + +internal delegate bool WorkflowEventHandler(TEvent evt) where TEvent : WorkflowEvent; +internal delegate ValueTask WorkflowEventHandlerAsync(TEvent evt) where TEvent : WorkflowEvent; + +internal class StreamingRunWatcher +{ + private readonly ConcurrentDictionary _eventHandlerChains = new(); + + public StreamingRunWatcher() + { + this.SuperStepCompleted += evt => + { + this.LastCheckpoint = evt.CompletionInfo?.Checkpoint; + return false; + }; + } + + public event WorkflowEventHandler WorkflowStarted + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler WorkflowError + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler WorkflowWarning + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler SuperStepStarted + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler SuperStepCompleted + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler ExecutorInvoked + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler ExecutorFailed + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler ExecutorCompleted + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler AgentRunUpdate + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler AgentRunResponse + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler RequestInfo + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler WorkflowOutput + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler SubworkflowError + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public event WorkflowEventHandler SubworkflowWarning + { + add => this.For().Event += value; + remove => this.For().Event -= value; + } + + public EventBinding For() where TEvent : WorkflowEvent + { + IWorkflowEventChain chainObj = this._eventHandlerChains.GetOrAdd(typeof(TEvent), _ => new EventChain()); + Debug.Assert(chainObj is EventChain); + EventChain eventChain = (EventChain)chainObj; + + return new(eventChain); + } + + public async ValueTask RunToHaltAsync(StreamingRun streamingRun, CancellationToken cancellationToken = default) + { + List pendingRequestInfoEvents = new(); + + // TODO: Should we make some kind of facility to make it easier to shunt events to run on the "UI" / "main" thread? + await foreach (WorkflowEvent evt in streamingRun.WatchStreamAsync(cancellationToken) + .WithCancellation(cancellationToken) + .ConfigureAwait(false)) + { + if (evt is RequestInfoEvent requestInfoEvt) + { + pendingRequestInfoEvents.Add(requestInfoEvt); + continue; + } + + await TryInvokeHandlerAsync(evt).ConfigureAwait(false); + + if (evt is SuperStepCompletedEvent) + { + foreach (RequestInfoEvent pendingEvt in pendingRequestInfoEvents) + { + await TryInvokeHandlerAsync(pendingEvt).ConfigureAwait(false); + } + + pendingRequestInfoEvents.Clear(); + } + } + + async ValueTask TryInvokeHandlerAsync(WorkflowEvent evt) + { + Type? eventType = evt.GetType(); + bool handled = false; + + while (!handled && eventType != null) + { + // Check if there are any handlers for this event type, and see if they handle the event. + if (this._eventHandlerChains.TryGetValue(evt.GetType(), out IWorkflowEventChain? chain)) + { + handled = await chain.RaiseAsync(evt, cancellationToken).ConfigureAwait(false); + } + + // If we reach the base WorkflowEvent type, stop looking further. + if (eventType == typeof(WorkflowEvent)) + { + break; + } + + eventType = eventType.BaseType; + } + } + } + + public CheckpointInfo? LastCheckpoint { get; private set; } +} + +internal static class Sample +{ + public static async ValueTask RunAsync() + { + WorkflowBuilder builder = new("start"); + // build the workflow + + Workflow workflow = builder.Build(); + + StreamingRunWatcher watcher = new(); + watcher.RequestInfo += (evt) => + { + // handle the request + return false; + }; + + watcher.AgentRunUpdate += (evt) => + { + Console.Write(evt.Update.Text); + return true; + }; + + watcher.For().Event += (evt) => + { + // log the event + return false; + }; + + StreamingRun streamingRun = await InProcessExecution.Concurrent + .StreamAsync(workflow, input: "some input") + .ConfigureAwait(false); + await watcher.RunToHaltAsync(streamingRun).ConfigureAwait(false); + } +}